opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix telegram media group album handling

Browse source
This commit is contained in:
Anton Bogdanovich 2026-05-03 20:40:20 -07:00
parent 6e6293e596
commit 09d3dff432
5 changed files with 437 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/channels/telegram/README.md
View file

@ -15,7 +15,8 @@ The Telegram channel uses long polling via the Telegram Bot API for bot-based co
"token": "123456789:ABCdefGHIjklMNOpqrsTUVwxyz",
"allow_from": ["123456789"],
"proxy": "",
"use_markdown_v2": false
"use_markdown_v2": false,
"media_group_delay_ms": 500
}
}
}
@ -28,6 +29,7 @@ The Telegram channel uses long polling via the Telegram Bot API for bot-based co
| allow_from | array | No | Allowlist of user IDs; empty means all users are allowed |
| proxy | string | No | Proxy URL for connecting to the Telegram API (e.g. http://127.0.0.1:7890) |
| use_markdown_v2 | bool | No | Enable Telegram MarkdownV2 formatting |
| media_group_delay_ms | int | No | Idle delay before processing Telegram media groups/albums. Defaults to 500 ms |
## Setup

295
pkg/channels/telegram/telegram.go
View file

@ -11,6 +11,7 @@ import (
"net/url"
"os"
"regexp"
"slices"
"strconv"
"strings"
"sync"
@ -43,20 +44,38 @@ var (
reInlineCode = regexp.MustCompile("`([^`]+)`")
)
const defaultMediaGroupDelay = 500 * time.Millisecond
type TelegramChannel struct {
*channels.BaseChannel
bot *telego.Bot
bh *th.BotHandler
bc *config.Channel
chatIDs map[string]int64
ctx context.Context
cancel context.CancelFunc
tgCfg *config.TelegramSettings
progress *channels.ToolFeedbackAnimator
bot *telego.Bot
bh *th.BotHandler
bc *config.Channel
chatIDsMu sync.Mutex
chatIDs map[string]int64
ctx context.Context
cancel context.CancelFunc
tgCfg *config.TelegramSettings
progress *channels.ToolFeedbackAnimator
registerFunc func(context.Context, []commands.Definition) error
commandRegDelayFn func(int) time.Duration
commandRegCancel context.CancelFunc
mediaGroupMu sync.Mutex
mediaGroups map[string]*telegramMediaGroup
mediaGroupDelay time.Duration
}
type telegramMediaGroup struct {
messages []*telego.Message
timer *time.Timer
generation uint64
}
type telegramMessageParts struct {
content []string
mediaPaths []string
}
func NewTelegramChannel(
@ -112,11 +131,21 @@ func NewTelegramChannel(
bc: bc,
chatIDs: make(map[string]int64),
tgCfg: telegramCfg,
mediaGroups: make(map[string]*telegramMediaGroup),
mediaGroupDelay: telegramMediaGroupDelay(telegramCfg),
}
ch.progress = channels.NewToolFeedbackAnimator(ch.EditMessage)
return ch, nil
}
func telegramMediaGroupDelay(telegramCfg *config.TelegramSettings) time.Duration {
if telegramCfg != nil && telegramCfg.MediaGroupDelayMS > 0 {
return time.Duration(telegramCfg.MediaGroupDelayMS) * time.Millisecond
}
return defaultMediaGroupDelay
}
func (c *TelegramChannel) Start(ctx context.Context) error {
logger.InfoC("telegram", "Starting Telegram bot (polling mode)...")
@ -167,6 +196,7 @@ func (c *TelegramChannel) Stop(ctx context.Context) error {
if c.bh != nil {
_ = c.bh.StopWithContext(ctx)
}
c.flushPendingMediaGroups(ctx)
// Cancel our context (stops long polling)
if c.cancel != nil {
@ -713,6 +743,131 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe
}
func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error {
if message != nil && strings.TrimSpace(message.MediaGroupID) != "" {
return c.bufferMediaGroupMessage(ctx, message)
}
return c.handleMessages(ctx, []*telego.Message{message})
}
func (c *TelegramChannel) bufferMediaGroupMessage(ctx context.Context, message *telego.Message) error {
if message == nil {
return fmt.Errorf("message is nil")
}
groupID := strings.TrimSpace(message.MediaGroupID)
if groupID == "" {
return c.handleMessages(ctx, []*telego.Message{message})
}
msgCopy := *message
msgCopy.Photo = append([]telego.PhotoSize(nil), message.Photo...)
key := fmt.Sprintf("%d:%s", message.Chat.ID, groupID)
c.mediaGroupMu.Lock()
if c.mediaGroups == nil {
c.mediaGroups = make(map[string]*telegramMediaGroup)
}
group := c.mediaGroups[key]
if group == nil {
group = &telegramMediaGroup{}
c.mediaGroups[key] = group
}
group.messages = append(group.messages, &msgCopy)
group.generation++
generation := group.generation
if group.timer != nil {
group.timer.Stop()
}
delay := c.mediaGroupDelay
if delay <= 0 {
delay = defaultMediaGroupDelay
}
group.timer = time.AfterFunc(delay, func() {
c.flushMediaGroup(c.ctx, key, generation)
})
c.mediaGroupMu.Unlock()
logger.DebugCF("telegram", "Buffered media group message", map[string]any{
"chat_id": message.Chat.ID,
"media_group_id": groupID,
"message_id": message.MessageID,
})
return nil
}
func (c *TelegramChannel) flushPendingMediaGroups(ctx context.Context) {
c.mediaGroupMu.Lock()
keys := make([]string, 0, len(c.mediaGroups))
for key, group := range c.mediaGroups {
if group.timer != nil {
group.timer.Stop()
}
keys = append(keys, key)
}
c.mediaGroupMu.Unlock()
for _, key := range keys {
c.flushMediaGroup(ctx, key, 0)
}
}
func (c *TelegramChannel) flushMediaGroup(ctx context.Context, key string, generation uint64) {
c.mediaGroupMu.Lock()
group := c.mediaGroups[key]
if group == nil {
c.mediaGroupMu.Unlock()
return
}
if generation != 0 && group.generation != generation {
c.mediaGroupMu.Unlock()
return
}
delete(c.mediaGroups, key)
if group.timer != nil {
group.timer.Stop()
}
messages := append([]*telego.Message(nil), group.messages...)
c.mediaGroupMu.Unlock()
if len(messages) == 0 {
return
}
slices.SortFunc(messages, func(a, b *telego.Message) int {
switch {
case a == nil && b == nil:
return 0
case a == nil:
return -1
case b == nil:
return 1
default:
return a.MessageID - b.MessageID
}
})
if ctx == nil {
ctx = context.Background()
}
if err := c.handleMessages(ctx, messages); err != nil {
logger.ErrorCF("telegram", "Failed to handle media group", map[string]any{
"key": key,
"error": err.Error(),
})
}
}
func (c *TelegramChannel) handleMessages(ctx context.Context, messages []*telego.Message) error {
if len(messages) == 0 {
return nil
}
message := messages[0]
for _, candidate := range messages {
if candidate == nil {
continue
}
if strings.TrimSpace(candidate.Text) != "" || strings.TrimSpace(candidate.Caption) != "" {
message = candidate
break
}
}
if message == nil {
return fmt.Errorf("message is nil")
}
@ -740,7 +895,9 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
}
chatID := message.Chat.ID
c.chatIDsMu.Lock()
c.chatIDs[platformID] = chatID
c.chatIDsMu.Unlock()
content := ""
mediaPaths := []string{}
@ -764,61 +921,18 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
return localPath // fallback: use raw path
}
if message.Text != "" {
content += message.Text
}
if message.Caption != "" {
if content != "" {
content += "\n"
for i, msg := range messages {
if msg == nil {
continue
}
content += message.Caption
}
if len(message.Photo) > 0 {
photo := message.Photo[len(message.Photo)-1]
photoPath := c.downloadPhoto(ctx, photo.FileID)
if photoPath != "" {
mediaPaths = append(mediaPaths, storeMedia(photoPath, "photo.jpg"))
parts := c.collectTelegramMessageParts(ctx, msg, i, len(messages), storeMedia)
for _, part := range parts.content {
if content != "" {
content += "\n"
}
content += "[image: photo]"
}
}
if message.Voice != nil {
voicePath := c.downloadFile(ctx, message.Voice.FileID, ".ogg")
if voicePath != "" {
mediaPaths = append(mediaPaths, storeMedia(voicePath, "voice.ogg"))
if content != "" {
content += "\n"
}
content += "[voice]"
}
}
if message.Audio != nil {
audioPath := c.downloadFile(ctx, message.Audio.FileID, ".mp3")
if audioPath != "" {
mediaPaths = append(mediaPaths, storeMedia(audioPath, "audio.mp3"))
if content != "" {
content += "\n"
}
content += "[audio]"
}
}
if message.Document != nil {
docPath := c.downloadFile(ctx, message.Document.FileID, "")
if docPath != "" {
mediaPaths = append(mediaPaths, storeMedia(docPath, "document"))
if content != "" {
content += "\n"
}
content += "[file]"
content += part
}
mediaPaths = append(mediaPaths, parts.mediaPaths...)
}
if content == "" && len(mediaPaths) == 0 {
@ -917,6 +1031,71 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
return nil
}
func (c *TelegramChannel) collectTelegramMessageParts(
ctx context.Context,
msg *telego.Message,
index int,
total int,
storeMedia func(localPath, filename string) string,
) telegramMessageParts {
parts := telegramMessageParts{}
if msg == nil {
return parts
}
if text := strings.TrimSpace(msg.Text); text != "" {
parts.content = append(parts.content, text)
}
if caption := strings.TrimSpace(msg.Caption); caption != "" {
parts.content = append(parts.content, caption)
}
if len(msg.Photo) > 0 {
photo := msg.Photo[len(msg.Photo)-1]
photoPath := c.downloadPhoto(ctx, photo.FileID)
if photoPath != "" {
photoNumber := index + 1
parts.mediaPaths = append(parts.mediaPaths, storeMedia(photoPath, fmt.Sprintf("photo-%d.jpg", photoNumber)))
parts.content = append(parts.content, fmt.Sprintf("[image: photo %d]", photoNumber))
}
}
if msg.Voice != nil {
voicePath := c.downloadFile(ctx, msg.Voice.FileID, ".ogg")
if voicePath != "" {
parts.mediaPaths = append(parts.mediaPaths, storeMedia(voicePath, indexedMediaFilename("voice", ".ogg", index, total)))
parts.content = append(parts.content, "[voice]")
}
}
if msg.Audio != nil {
audioPath := c.downloadFile(ctx, msg.Audio.FileID, ".mp3")
if audioPath != "" {
filename := msg.Audio.FileName
if strings.TrimSpace(filename) == "" {
filename = indexedMediaFilename("audio", ".mp3", index, total)
}
parts.mediaPaths = append(parts.mediaPaths, storeMedia(audioPath, filename))
parts.content = append(parts.content, "[audio]")
}
}
if msg.Document != nil {
docPath := c.downloadFile(ctx, msg.Document.FileID, "")
if docPath != "" {
filename := msg.Document.FileName
if strings.TrimSpace(filename) == "" {
filename = indexedMediaFilename("document", "", index, total)
}
parts.mediaPaths = append(parts.mediaPaths, storeMedia(docPath, filename))
parts.content = append(parts.content, "[file]")
}
}
return parts
}
func indexedMediaFilename(prefix, ext string, index int, total int) string {
if total <= 1 {
return prefix + ext
}
return fmt.Sprintf("%s-%d%s", prefix, index+1, ext)
}
func (c *TelegramChannel) prependTelegramQuotedReply(content string, reply *telego.Message) string {
quoted := strings.TrimSpace(telegramQuotedContent(reply))
if quoted == "" {

188
pkg/channels/telegram/telegram_test.go
View file

@ -10,6 +10,7 @@ import (
"strconv"
"strings"
"testing"
"time"
"github.com/mymmrac/telego"
ta "github.com/mymmrac/telego/telegoapi"
@ -1100,3 +1101,190 @@ func TestHandleMessage_EmptyContent_Ignored(t *testing.T) {
default:
}
}
func TestHandleMessage_MediaGroupCombinesCaptionMessages(t *testing.T) {
messageBus, ch := newMediaGroupTestChannel(10 * time.Millisecond)
base := testMediaGroupMessage("album-1")
first := base
first.MessageID = 1
second := base
second.MessageID = 2
second.Caption = "meal caption"
require.NoError(t, ch.handleMessage(context.Background(), &first))
require.NoError(t, ch.handleMessage(context.Background(), &second))
select {
case inbound := <-messageBus.InboundChan():
assert.Equal(t, "2", inbound.Context.MessageID)
assert.Equal(t, "meal caption", inbound.Content)
case <-time.After(time.Second):
t.Fatal("timed out waiting for combined media group message")
}
}
func TestHandleMessage_MediaGroupWaitsForStaggeredMessages(t *testing.T) {
messageBus, ch := newMediaGroupTestChannel(100 * time.Millisecond)
base := testMediaGroupMessage("album-staggered")
first := base
first.MessageID = 1
first.Caption = "first caption"
second := base
second.MessageID = 2
second.Caption = "second caption"
require.NoError(t, ch.handleMessage(context.Background(), &first))
time.Sleep(50 * time.Millisecond)
require.NoError(t, ch.handleMessage(context.Background(), &second))
select {
case inbound := <-messageBus.InboundChan():
t.Fatalf("media group flushed before idle delay reset: %#v", inbound)
case <-time.After(75 * time.Millisecond):
}
select {
case inbound := <-messageBus.InboundChan():
assert.Equal(t, "1", inbound.Context.MessageID)
assert.Equal(t, "first caption\nsecond caption", inbound.Content)
case <-time.After(time.Second):
t.Fatal("timed out waiting for staggered media group message")
}
}
func TestFlushMediaGroupIgnoresStaleTimerGeneration(t *testing.T) {
messageBus, ch := newMediaGroupTestChannel(time.Hour)
base := testMediaGroupMessage("album-generation")
first := base
first.MessageID = 1
first.Caption = "first"
second := base
second.MessageID = 2
second.Caption = "second"
key := "456:album-generation"
ch.mediaGroupMu.Lock()
ch.mediaGroups[key] = &telegramMediaGroup{
messages: []*telego.Message{&first, &second},
generation: 2,
}
ch.mediaGroupMu.Unlock()
ch.flushMediaGroup(context.Background(), key, 1)
select {
case inbound := <-messageBus.InboundChan():
t.Fatalf("stale media group generation flushed unexpectedly: %#v", inbound)
default:
}
ch.mediaGroupMu.Lock()
_, stillPending := ch.mediaGroups[key]
ch.mediaGroupMu.Unlock()
require.True(t, stillPending, "stale flush should leave the current batch pending")
ch.flushMediaGroup(context.Background(), key, 2)
select {
case inbound := <-messageBus.InboundChan():
assert.Equal(t, "1", inbound.Context.MessageID)
assert.Equal(t, "first\nsecond", inbound.Content)
case <-time.After(time.Second):
t.Fatal("timed out waiting for current generation media group flush")
}
}
func TestHandleMessage_MediaGroupAfterDelayStartsNewBatch(t *testing.T) {
messageBus, ch := newMediaGroupTestChannel(10 * time.Millisecond)
base := testMediaGroupMessage("album-split")
first := base
first.MessageID = 1
first.Caption = "first"
second := base
second.MessageID = 2
second.Caption = "second"
require.NoError(t, ch.handleMessage(context.Background(), &first))
select {
case inbound := <-messageBus.InboundChan():
assert.Equal(t, "1", inbound.Context.MessageID)
assert.Equal(t, "first", inbound.Content)
case <-time.After(time.Second):
t.Fatal("timed out waiting for first media group batch")
}
require.NoError(t, ch.handleMessage(context.Background(), &second))
select {
case inbound := <-messageBus.InboundChan():
assert.Equal(t, "2", inbound.Context.MessageID)
assert.Equal(t, "second", inbound.Content)
case <-time.After(time.Second):
t.Fatal("timed out waiting for second media group batch")
}
}
func TestStopFlushesPendingMediaGroups(t *testing.T) {
messageBus, ch := newMediaGroupTestChannel(time.Hour)
base := testMediaGroupMessage("album-stop")
msg := base
msg.MessageID = 1
msg.Caption = "caption before stop"
require.NoError(t, ch.handleMessage(context.Background(), &msg))
require.NoError(t, ch.Stop(context.Background()))
select {
case inbound := <-messageBus.InboundChan():
assert.Equal(t, "1", inbound.Context.MessageID)
assert.Equal(t, "caption before stop", inbound.Content)
case <-time.After(time.Second):
t.Fatal("timed out waiting for pending media group flush on stop")
}
}
func TestNewTelegramChannelUsesConfiguredMediaGroupDelay(t *testing.T) {
ch, err := NewTelegramChannel(
&config.Channel{Type: config.ChannelTelegram, Enabled: true},
&config.TelegramSettings{
Token: *config.NewSecureString(testToken),
MediaGroupDelayMS: 750,
},
bus.NewMessageBus(),
)
require.NoError(t, err)
assert.Equal(t, 750*time.Millisecond, ch.mediaGroupDelay)
ch, err = NewTelegramChannel(
&config.Channel{Type: config.ChannelTelegram, Enabled: true},
&config.TelegramSettings{Token: *config.NewSecureString(testToken)},
bus.NewMessageBus(),
)
require.NoError(t, err)
assert.Equal(t, defaultMediaGroupDelay, ch.mediaGroupDelay)
}
func newMediaGroupTestChannel(delay time.Duration) (*bus.MessageBus, *TelegramChannel) {
messageBus := bus.NewMessageBus()
ch := &TelegramChannel{
BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
chatIDs: make(map[string]int64),
ctx: context.Background(),
mediaGroups: make(map[string]*telegramMediaGroup),
mediaGroupDelay: delay,
}
return messageBus, ch
}
func testMediaGroupMessage(mediaGroupID string) telego.Message {
return telego.Message{
Chat: telego.Chat{
ID: 456,
Type: "private",
},
From: &telego.User{
ID: 789,
FirstName: "User",
},
MediaGroupID: mediaGroupID,
}
}

11
pkg/config/config.go
View file

@ -359,11 +359,12 @@ type WhatsAppSettings struct {
}
type TelegramSettings struct {
Token SecureString `json:"token,omitzero" yaml:"token,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"`
BaseURL string `json:"base_url" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_BASE_URL"`
Proxy string `json:"proxy" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"`
Streaming StreamingConfig `json:"streaming,omitempty" yaml:"-"`
UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_USE_MARKDOWN_V2"`
Token SecureString `json:"token,omitzero" yaml:"token,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"`
BaseURL string `json:"base_url" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_BASE_URL"`
Proxy string `json:"proxy" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"`
Streaming StreamingConfig `json:"streaming,omitempty" yaml:"-"`
UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_USE_MARKDOWN_V2"`
MediaGroupDelayMS int `json:"media_group_delay_ms" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_MEDIA_GROUP_DELAY_MS"`
}
type FeishuSettings struct {

5
pkg/config/defaults.go
View file

@ -496,8 +496,9 @@ func defaultChannels() ChannelsConfig {
"typing": map[string]any{"enabled": true},
"placeholder": map[string]any{"enabled": true, "text": []string{"Thinking... 💭"}},
"settings": map[string]any{
"streaming": map[string]any{"enabled": true, "throttle_seconds": 3, "min_growth_chars": 200},
"use_markdown_v2": false,
"streaming": map[string]any{"enabled": true, "throttle_seconds": 3, "min_growth_chars": 200},
"use_markdown_v2": false,
"media_group_delay_ms": 500,
},
},
"feishu": map[string]any{},