Files
picoclaw/pkg/channels/telegram/telegram.go
T
Hoshina 90b4a64683 feat(channels): add typing/placeholder automation and Pico Protocol channel (Phase 10 + 7)
Phase 10: Define TypingCapable, MessageEditor, PlaceholderRecorder interfaces.
Manager orchestrates outbound typing stop and placeholder editing via preSend.
Migrate Telegram, Discord, Slack, OneBot to register state with Manager instead
of handling locally in Send. Phase 7: Add native WebSocket Pico Protocol channel
as reference implementation of all optional capability interfaces.
2026-02-24 12:10:45 +08:00

664 lines
16 KiB
Go

package telegram
import (
"context"
"fmt"
"net/http"
"net/url"
"os"
"regexp"
"strconv"
"strings"
"time"
"github.com/mymmrac/telego"
"github.com/mymmrac/telego/telegohandler"
th "github.com/mymmrac/telego/telegohandler"
tu "github.com/mymmrac/telego/telegoutil"
"github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/channels"
"github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/media"
"github.com/sipeed/picoclaw/pkg/utils"
)
type TelegramChannel struct {
*channels.BaseChannel
bot *telego.Bot
bh *telegohandler.BotHandler
commands TelegramCommander
config *config.Config
chatIDs map[string]int64
ctx context.Context
cancel context.CancelFunc
}
func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChannel, error) {
var opts []telego.BotOption
telegramCfg := cfg.Channels.Telegram
if telegramCfg.Proxy != "" {
proxyURL, parseErr := url.Parse(telegramCfg.Proxy)
if parseErr != nil {
return nil, fmt.Errorf("invalid proxy URL %q: %w", telegramCfg.Proxy, parseErr)
}
opts = append(opts, telego.WithHTTPClient(&http.Client{
Transport: &http.Transport{
Proxy: http.ProxyURL(proxyURL),
},
}))
} else if os.Getenv("HTTP_PROXY") != "" || os.Getenv("HTTPS_PROXY") != "" {
// Use environment proxy if configured
opts = append(opts, telego.WithHTTPClient(&http.Client{
Transport: &http.Transport{
Proxy: http.ProxyFromEnvironment,
},
}))
}
bot, err := telego.NewBot(telegramCfg.Token, opts...)
if err != nil {
return nil, fmt.Errorf("failed to create telegram bot: %w", err)
}
base := channels.NewBaseChannel(
"telegram",
telegramCfg,
bus,
telegramCfg.AllowFrom,
channels.WithMaxMessageLength(4096),
channels.WithGroupTrigger(telegramCfg.GroupTrigger),
)
return &TelegramChannel{
BaseChannel: base,
commands: NewTelegramCommands(bot, cfg),
bot: bot,
config: cfg,
chatIDs: make(map[string]int64),
}, nil
}
func (c *TelegramChannel) Start(ctx context.Context) error {
logger.InfoC("telegram", "Starting Telegram bot (polling mode)...")
c.ctx, c.cancel = context.WithCancel(ctx)
updates, err := c.bot.UpdatesViaLongPolling(c.ctx, &telego.GetUpdatesParams{
Timeout: 30,
})
if err != nil {
c.cancel()
return fmt.Errorf("failed to start long polling: %w", err)
}
bh, err := telegohandler.NewBotHandler(c.bot, updates)
if err != nil {
c.cancel()
return fmt.Errorf("failed to create bot handler: %w", err)
}
c.bh = bh
bh.HandleMessage(func(ctx *th.Context, message telego.Message) error {
c.commands.Help(ctx, message)
return nil
}, th.CommandEqual("help"))
bh.HandleMessage(func(ctx *th.Context, message telego.Message) error {
return c.commands.Start(ctx, message)
}, th.CommandEqual("start"))
bh.HandleMessage(func(ctx *th.Context, message telego.Message) error {
return c.commands.Show(ctx, message)
}, th.CommandEqual("show"))
bh.HandleMessage(func(ctx *th.Context, message telego.Message) error {
return c.commands.List(ctx, message)
}, th.CommandEqual("list"))
bh.HandleMessage(func(ctx *th.Context, message telego.Message) error {
return c.handleMessage(ctx, &message)
}, th.AnyMessage())
c.SetRunning(true)
logger.InfoCF("telegram", "Telegram bot connected", map[string]any{
"username": c.bot.Username(),
})
go bh.Start()
return nil
}
func (c *TelegramChannel) Stop(ctx context.Context) error {
logger.InfoC("telegram", "Stopping Telegram bot...")
c.SetRunning(false)
// Stop the bot handler
if c.bh != nil {
c.bh.Stop()
}
// Cancel our context (stops long polling)
if c.cancel != nil {
c.cancel()
}
return nil
}
func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
if !c.IsRunning() {
return channels.ErrNotRunning
}
chatID, err := parseChatID(msg.ChatID)
if err != nil {
return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed)
}
htmlContent := markdownToTelegramHTML(msg.Content)
// Typing/placeholder handled by Manager.preSend — just send the message
tgMsg := tu.Message(tu.ID(chatID), htmlContent)
tgMsg.ParseMode = telego.ModeHTML
if _, err = c.bot.SendMessage(ctx, tgMsg); err != nil {
logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{
"error": err.Error(),
})
tgMsg.ParseMode = ""
if _, err = c.bot.SendMessage(ctx, tgMsg); err != nil {
return fmt.Errorf("telegram send: %w", channels.ErrTemporary)
}
}
return nil
}
// EditMessage implements channels.MessageEditor.
func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error {
cid, err := parseChatID(chatID)
if err != nil {
return err
}
mid, err := strconv.Atoi(messageID)
if err != nil {
return err
}
htmlContent := markdownToTelegramHTML(content)
editMsg := tu.EditMessageText(tu.ID(cid), mid, htmlContent)
editMsg.ParseMode = telego.ModeHTML
_, err = c.bot.EditMessageText(ctx, editMsg)
return err
}
// SendMedia implements the channels.MediaSender interface.
func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error {
if !c.IsRunning() {
return channels.ErrNotRunning
}
chatID, err := parseChatID(msg.ChatID)
if err != nil {
return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed)
}
store := c.GetMediaStore()
if store == nil {
return fmt.Errorf("no media store available: %w", channels.ErrSendFailed)
}
for _, part := range msg.Parts {
localPath, err := store.Resolve(part.Ref)
if err != nil {
logger.ErrorCF("telegram", "Failed to resolve media ref", map[string]any{
"ref": part.Ref,
"error": err.Error(),
})
continue
}
file, err := os.Open(localPath)
if err != nil {
logger.ErrorCF("telegram", "Failed to open media file", map[string]any{
"path": localPath,
"error": err.Error(),
})
continue
}
filename := part.Filename
if filename == "" {
filename = "file"
}
switch part.Type {
case "image":
params := &telego.SendPhotoParams{
ChatID: tu.ID(chatID),
Photo: telego.InputFile{File: file},
Caption: part.Caption,
}
_, err = c.bot.SendPhoto(ctx, params)
case "audio":
params := &telego.SendAudioParams{
ChatID: tu.ID(chatID),
Audio: telego.InputFile{File: file},
Caption: part.Caption,
}
_, err = c.bot.SendAudio(ctx, params)
case "video":
params := &telego.SendVideoParams{
ChatID: tu.ID(chatID),
Video: telego.InputFile{File: file},
Caption: part.Caption,
}
_, err = c.bot.SendVideo(ctx, params)
default: // "file" or unknown types
params := &telego.SendDocumentParams{
ChatID: tu.ID(chatID),
Document: telego.InputFile{File: file},
Caption: part.Caption,
}
_, err = c.bot.SendDocument(ctx, params)
}
file.Close()
if err != nil {
logger.ErrorCF("telegram", "Failed to send media", map[string]any{
"type": part.Type,
"error": err.Error(),
})
return fmt.Errorf("telegram send media: %w", channels.ErrTemporary)
}
}
return nil
}
func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error {
if message == nil {
return fmt.Errorf("message is nil")
}
user := message.From
if user == nil {
return fmt.Errorf("message sender (user) is nil")
}
senderID := fmt.Sprintf("%d", user.ID)
if user.Username != "" {
senderID = fmt.Sprintf("%d|%s", user.ID, user.Username)
}
// 检查白名单,避免为被拒绝的用户下载附件
if !c.IsAllowed(senderID) {
logger.DebugCF("telegram", "Message rejected by allowlist", map[string]any{
"user_id": senderID,
})
return nil
}
chatID := message.Chat.ID
c.chatIDs[senderID] = chatID
content := ""
mediaPaths := []string{}
chatIDStr := fmt.Sprintf("%d", chatID)
messageIDStr := fmt.Sprintf("%d", message.MessageID)
scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr)
// Helper to register a local file with the media store
storeMedia := func(localPath, filename string) string {
if store := c.GetMediaStore(); store != nil {
ref, err := store.Store(localPath, media.MediaMeta{
Filename: filename,
Source: "telegram",
}, scope)
if err == nil {
return ref
}
}
return localPath // fallback: use raw path
}
if message.Text != "" {
content += message.Text
}
if message.Caption != "" {
if content != "" {
content += "\n"
}
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"))
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]"
}
}
if content == "" {
content = "[empty message]"
}
// In group chats, apply unified group trigger filtering
if message.Chat.Type != "private" {
isMentioned := c.isBotMentioned(message)
if isMentioned {
content = c.stripBotMention(content)
}
respond, cleaned := c.ShouldRespondInGroup(isMentioned, content)
if !respond {
return nil
}
content = cleaned
}
logger.DebugCF("telegram", "Received message", map[string]any{
"sender_id": senderID,
"chat_id": fmt.Sprintf("%d", chatID),
"preview": utils.Truncate(content, 50),
})
// Thinking indicator
err := c.bot.SendChatAction(ctx, tu.ChatAction(tu.ID(chatID), telego.ChatActionTyping))
if err != nil {
logger.ErrorCF("telegram", "Failed to send chat action", map[string]any{
"error": err.Error(),
})
}
// Create cancel function for thinking state and register with Manager
_, thinkCancel := context.WithTimeout(ctx, 5*time.Minute)
if rec := c.GetPlaceholderRecorder(); rec != nil {
rec.RecordTypingStop("telegram", chatIDStr, thinkCancel)
} else {
// No recorder — cancel immediately to avoid context leak
thinkCancel()
}
pMsg, err := c.bot.SendMessage(ctx, tu.Message(tu.ID(chatID), "Thinking... 💭"))
if err == nil {
pID := pMsg.MessageID
if rec := c.GetPlaceholderRecorder(); rec != nil {
rec.RecordPlaceholder("telegram", chatIDStr, fmt.Sprintf("%d", pID))
}
}
peerKind := "direct"
peerID := fmt.Sprintf("%d", user.ID)
if message.Chat.Type != "private" {
peerKind = "group"
peerID = fmt.Sprintf("%d", chatID)
}
peer := bus.Peer{Kind: peerKind, ID: peerID}
messageID := fmt.Sprintf("%d", message.MessageID)
metadata := map[string]string{
"user_id": fmt.Sprintf("%d", user.ID),
"username": user.Username,
"first_name": user.FirstName,
"is_group": fmt.Sprintf("%t", message.Chat.Type != "private"),
}
c.HandleMessage(
peer,
messageID,
fmt.Sprintf("%d", user.ID),
fmt.Sprintf("%d", chatID),
content,
mediaPaths,
metadata,
)
return nil
}
func (c *TelegramChannel) downloadPhoto(ctx context.Context, fileID string) string {
file, err := c.bot.GetFile(ctx, &telego.GetFileParams{FileID: fileID})
if err != nil {
logger.ErrorCF("telegram", "Failed to get photo file", map[string]any{
"error": err.Error(),
})
return ""
}
return c.downloadFileWithInfo(file, ".jpg")
}
func (c *TelegramChannel) downloadFileWithInfo(file *telego.File, ext string) string {
if file.FilePath == "" {
return ""
}
url := c.bot.FileDownloadURL(file.FilePath)
logger.DebugCF("telegram", "File URL", map[string]any{"url": url})
// Use FilePath as filename for better identification
filename := file.FilePath + ext
return utils.DownloadFile(url, filename, utils.DownloadOptions{
LoggerPrefix: "telegram",
})
}
func (c *TelegramChannel) downloadFile(ctx context.Context, fileID, ext string) string {
file, err := c.bot.GetFile(ctx, &telego.GetFileParams{FileID: fileID})
if err != nil {
logger.ErrorCF("telegram", "Failed to get file", map[string]any{
"error": err.Error(),
})
return ""
}
return c.downloadFileWithInfo(file, ext)
}
func parseChatID(chatIDStr string) (int64, error) {
var id int64
_, err := fmt.Sscanf(chatIDStr, "%d", &id)
return id, err
}
func markdownToTelegramHTML(text string) string {
if text == "" {
return ""
}
codeBlocks := extractCodeBlocks(text)
text = codeBlocks.text
inlineCodes := extractInlineCodes(text)
text = inlineCodes.text
text = regexp.MustCompile(`^#{1,6}\s+(.+)$`).ReplaceAllString(text, "$1")
text = regexp.MustCompile(`^>\s*(.*)$`).ReplaceAllString(text, "$1")
text = escapeHTML(text)
text = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`).ReplaceAllString(text, `<a href="$2">$1</a>`)
text = regexp.MustCompile(`\*\*(.+?)\*\*`).ReplaceAllString(text, "<b>$1</b>")
text = regexp.MustCompile(`__(.+?)__`).ReplaceAllString(text, "<b>$1</b>")
reItalic := regexp.MustCompile(`_([^_]+)_`)
text = reItalic.ReplaceAllStringFunc(text, func(s string) string {
match := reItalic.FindStringSubmatch(s)
if len(match) < 2 {
return s
}
return "<i>" + match[1] + "</i>"
})
text = regexp.MustCompile(`~~(.+?)~~`).ReplaceAllString(text, "<s>$1</s>")
text = regexp.MustCompile(`^[-*]\s+`).ReplaceAllString(text, "• ")
for i, code := range inlineCodes.codes {
escaped := escapeHTML(code)
text = strings.ReplaceAll(text, fmt.Sprintf("\x00IC%d\x00", i), fmt.Sprintf("<code>%s</code>", escaped))
}
for i, code := range codeBlocks.codes {
escaped := escapeHTML(code)
text = strings.ReplaceAll(
text,
fmt.Sprintf("\x00CB%d\x00", i),
fmt.Sprintf("<pre><code>%s</code></pre>", escaped),
)
}
return text
}
type codeBlockMatch struct {
text string
codes []string
}
func extractCodeBlocks(text string) codeBlockMatch {
re := regexp.MustCompile("```[\\w]*\\n?([\\s\\S]*?)```")
matches := re.FindAllStringSubmatch(text, -1)
codes := make([]string, 0, len(matches))
for _, match := range matches {
codes = append(codes, match[1])
}
i := 0
text = re.ReplaceAllStringFunc(text, func(m string) string {
placeholder := fmt.Sprintf("\x00CB%d\x00", i)
i++
return placeholder
})
return codeBlockMatch{text: text, codes: codes}
}
type inlineCodeMatch struct {
text string
codes []string
}
func extractInlineCodes(text string) inlineCodeMatch {
re := regexp.MustCompile("`([^`]+)`")
matches := re.FindAllStringSubmatch(text, -1)
codes := make([]string, 0, len(matches))
for _, match := range matches {
codes = append(codes, match[1])
}
i := 0
text = re.ReplaceAllStringFunc(text, func(m string) string {
placeholder := fmt.Sprintf("\x00IC%d\x00", i)
i++
return placeholder
})
return inlineCodeMatch{text: text, codes: codes}
}
func escapeHTML(text string) string {
text = strings.ReplaceAll(text, "&", "&amp;")
text = strings.ReplaceAll(text, "<", "&lt;")
text = strings.ReplaceAll(text, ">", "&gt;")
return text
}
// isBotMentioned checks if the bot is mentioned in the message via entities.
func (c *TelegramChannel) isBotMentioned(message *telego.Message) bool {
botUsername := c.bot.Username()
if botUsername == "" {
return false
}
entities := message.Entities
if entities == nil {
entities = message.CaptionEntities
}
for _, entity := range entities {
if entity.Type == "mention" {
// Extract the mention text from the message
text := message.Text
if text == "" {
text = message.Caption
}
runes := []rune(text)
end := entity.Offset + entity.Length
if end <= len(runes) {
mention := string(runes[entity.Offset:end])
if strings.EqualFold(mention, "@"+botUsername) {
return true
}
}
}
if entity.Type == "text_mention" && entity.User != nil {
if entity.User.Username == botUsername {
return true
}
}
}
return false
}
// stripBotMention removes the @bot mention from the content.
func (c *TelegramChannel) stripBotMention(content string) string {
botUsername := c.bot.Username()
if botUsername == "" {
return content
}
// Case-insensitive replacement
re := regexp.MustCompile(`(?i)@` + regexp.QuoteMeta(botUsername))
content = re.ReplaceAllString(content, "")
return strings.TrimSpace(content)
}