DBT/clawgo
1
0
Fork 0
mirror of https://github.com/YspCoder/clawgo.git synced 2026-04-15 03:57:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add interactive buttons support to telegram channel

Browse Source
This commit is contained in:
root
2026-02-14 16:06:36 +00:00
parent b3b7c3edfd
commit f56005246d
2 changed files with 95 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
pkg/bus/types.go
View File

@@ -10,10 +10,16 @@ type InboundMessage struct {
Metadata map[string]string `json:"metadata,omitempty"` Metadata map[string]string `json:"metadata,omitempty"`
} }
type Button struct {
Text string `json:"text"`
Data string `json:"data"`
}
type OutboundMessage struct { type OutboundMessage struct {
Channel string `json:"channel"` Channel string `json:"channel"`
ChatID string `json:"chat_id"` ChatID string `json:"chat_id"`
Content string `json:"content"` Content string `json:"content"`
Buttons [][]Button `json:"buttons,omitempty"`
} }
type MessageHandler func(InboundMessage) error type MessageHandler func(InboundMessage) error

206
pkg/channels/telegram.go
View File

@@ -133,6 +133,8 @@ func (c *TelegramChannel) Start(ctx context.Context) error {
} }
if update.Message != nil { if update.Message != nil {
c.dispatchHandleMessage(runCtx, update.Message) c.dispatchHandleMessage(runCtx, update.Message)
} else if update.CallbackQuery != nil {
c.handleCallbackQuery(runCtx, update.CallbackQuery)
} }
} }
} }
@@ -170,10 +172,6 @@ func (c *TelegramChannel) Stop(ctx context.Context) error {
return true return true
}) })
// In telego v1.x, the long polling is stopped by canceling the context
// passed to UpdatesViaLongPolling. We don't need a separate Stop call
// if we use the parent context correctly.
return nil return nil
} }
@@ -203,6 +201,35 @@ func (c *TelegramChannel) dispatchHandleMessage(runCtx context.Context, message
}(message) }(message)
} }
func (c *TelegramChannel) handleCallbackQuery(ctx context.Context, query *telego.CallbackQuery) {
if query == nil || query.Message == nil {
return
}
senderID := fmt.Sprintf("%d", query.From.ID)
chatID := fmt.Sprintf("%d", query.Message.GetChat().ID)
answerCtx, cancel := withTelegramAPITimeout(ctx)
_ = c.bot.AnswerCallbackQuery(answerCtx, &telego.AnswerCallbackQueryParams{
CallbackQueryID: query.ID,
})
cancel()
logger.InfoCF("telegram", "Callback query received", map[string]interface{}{
"sender_id": senderID,
"data": query.Data,
})
if !c.IsAllowed(senderID) {
return
}
c.HandleMessage(senderID, chatID, query.Data, nil, map[string]string{
"is_callback": "true",
"callback_id": query.ID,
})
}
func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
if !c.IsRunning() { if !c.IsRunning() {
return fmt.Errorf("telegram bot not running") return fmt.Errorf("telegram bot not running")
@@ -214,57 +241,54 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
} }
chatID := telegoutil.ID(chatIDInt) chatID := telegoutil.ID(chatIDInt)
// Stop thinking animation first to avoid animation/update races.
if stop, ok := c.stopThinking.LoadAndDelete(msg.ChatID); ok { if stop, ok := c.stopThinking.LoadAndDelete(msg.ChatID); ok {
logger.DebugCF("telegram", "Telegram thinking stop signal", map[string]interface{}{
logger.FieldChatID: msg.ChatID,
})
safeCloseSignal(stop) safeCloseSignal(stop)
} else {
logger.DebugCF("telegram", "Telegram thinking stop skipped (not found)", map[string]interface{}{
logger.FieldChatID: msg.ChatID,
})
} }
htmlContent := sanitizeTelegramHTML(markdownToTelegramHTML(msg.Content)) htmlContent := sanitizeTelegramHTML(markdownToTelegramHTML(msg.Content))
// Try to edit placeholder var markup *telego.InlineKeyboardMarkup
if pID, ok := c.placeholders.Load(msg.ChatID); ok { if len(msg.Buttons) > 0 {
// Always reset placeholder state even when edit/send fails. var rows [][]telego.InlineKeyboardButton
defer c.placeholders.Delete(msg.ChatID) for _, row := range msg.Buttons {
logger.DebugCF("telegram", "Telegram editing thinking placeholder", map[string]interface{}{ var buttons []telego.InlineKeyboardButton
logger.FieldChatID: msg.ChatID, for _, btn := range row {
"message_id": pID.(int), buttons = append(buttons, telegoutil.InlineKeyboardButton(btn.Text).WithCallbackData(btn.Data))
}) }
rows = append(rows, buttons)
}
markup = telegoutil.InlineKeyboard(rows...)
}
if pID, ok := c.placeholders.Load(msg.ChatID); ok {
defer c.placeholders.Delete(msg.ChatID)
editCtx, cancelEdit := withTelegramAPITimeout(ctx) editCtx, cancelEdit := withTelegramAPITimeout(ctx)
_, err := c.bot.EditMessageText(editCtx, &telego.EditMessageTextParams{ params := &telego.EditMessageTextParams{
ChatID: chatID, ChatID: chatID,
MessageID: pID.(int), MessageID: pID.(int),
Text: htmlContent, Text: htmlContent,
ParseMode: telego.ModeHTML, ParseMode: telego.ModeHTML,
}) ReplyMarkup: markup,
}
_, err := c.bot.EditMessageText(editCtx, params)
cancelEdit() cancelEdit()
if err == nil { if err == nil {
logger.DebugCF("telegram", "Telegram placeholder updated", map[string]interface{}{
logger.FieldChatID: msg.ChatID,
})
return nil return nil
} }
logger.WarnCF("telegram", "Telegram placeholder update failed; fallback to new message", map[string]interface{}{ logger.WarnCF("telegram", "Placeholder update failed; fallback to new message", map[string]interface{}{
logger.FieldChatID: msg.ChatID, logger.FieldChatID: msg.ChatID,
logger.FieldError: err.Error(), logger.FieldError: err.Error(),
}) })
// Fallback to new message if edit fails }
} else {
logger.DebugCF("telegram", "Telegram placeholder not found, sending new message", map[string]interface{}{ sendParams := telegoutil.Message(chatID, htmlContent).WithParseMode(telego.ModeHTML)
logger.FieldChatID: msg.ChatID, if markup != nil {
}) sendParams.WithReplyMarkup(markup)
} }
sendCtx, cancelSend := withTelegramAPITimeout(ctx) sendCtx, cancelSend := withTelegramAPITimeout(ctx)
_, err = c.bot.SendMessage(sendCtx, telegoutil.Message(chatID, htmlContent).WithParseMode(telego.ModeHTML)) _, err = c.bot.SendMessage(sendCtx, sendParams)
cancelSend() cancelSend()
if err != nil { if err != nil {
@@ -272,15 +296,13 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
logger.FieldError: err.Error(), logger.FieldError: err.Error(),
}) })
plain := plainTextFromTelegramHTML(htmlContent) plain := plainTextFromTelegramHTML(htmlContent)
sendPlainCtx, cancelSendPlain := withTelegramAPITimeout(ctx) sendPlainParams := telegoutil.Message(chatID, plain)
_, err = c.bot.SendMessage(sendPlainCtx, telegoutil.Message(chatID, plain)) if markup != nil {
cancelSendPlain() sendPlainParams.WithReplyMarkup(markup)
if err != nil {
logger.ErrorCF("telegram", "Telegram plain-text fallback send failed", map[string]interface{}{
logger.FieldChatID: msg.ChatID,
logger.FieldError: err.Error(),
})
} }
sendPlainCtx, cancelSendPlain := withTelegramAPITimeout(ctx)
_, err = c.bot.SendMessage(sendPlainCtx, sendPlainParams)
cancelSendPlain()
return err return err
} }
@@ -402,7 +424,6 @@ func (c *TelegramChannel) handleMessage(runCtx context.Context, message *telego.
return return
} }
// Thinking indicator
apiCtx, cancelAPI := context.WithTimeout(runCtx, telegramAPICallTimeout) apiCtx, cancelAPI := context.WithTimeout(runCtx, telegramAPICallTimeout)
_ = c.bot.SendChatAction(apiCtx, &telego.SendChatActionParams{ _ = c.bot.SendChatAction(apiCtx, &telego.SendChatActionParams{
ChatID: telegoutil.ID(chatID), ChatID: telegoutil.ID(chatID),
@@ -412,13 +433,9 @@ func (c *TelegramChannel) handleMessage(runCtx context.Context, message *telego.
stopChan := make(chan struct{}) stopChan := make(chan struct{})
if prev, ok := c.stopThinking.LoadAndDelete(fmt.Sprintf("%d", chatID)); ok { if prev, ok := c.stopThinking.LoadAndDelete(fmt.Sprintf("%d", chatID)); ok {
// Ensure previous animation loop exits before replacing channel.
safeCloseSignal(prev) safeCloseSignal(prev)
} }
c.stopThinking.Store(fmt.Sprintf("%d", chatID), stopChan) c.stopThinking.Store(fmt.Sprintf("%d", chatID), stopChan)
logger.DebugCF("telegram", "Telegram thinking started", map[string]interface{}{
logger.FieldChatID: chatID,
})
sendCtx, cancelSend := context.WithTimeout(runCtx, telegramAPICallTimeout) sendCtx, cancelSend := context.WithTimeout(runCtx, telegramAPICallTimeout)
pMsg, err := c.bot.SendMessage(sendCtx, telegoutil.Message(telegoutil.ID(chatID), "Thinking... 💭")) pMsg, err := c.bot.SendMessage(sendCtx, telegoutil.Message(telegoutil.ID(chatID), "Thinking... 💭"))
@@ -426,10 +443,6 @@ func (c *TelegramChannel) handleMessage(runCtx context.Context, message *telego.
if err == nil { if err == nil {
pID := pMsg.MessageID pID := pMsg.MessageID
c.placeholders.Store(fmt.Sprintf("%d", chatID), pID) c.placeholders.Store(fmt.Sprintf("%d", chatID), pID)
logger.DebugCF("telegram", "Telegram thinking placeholder created", map[string]interface{}{
logger.FieldChatID: chatID,
"message_id": pID,
})
go func(cid int64, mid int, stop <-chan struct{}, parentCtx context.Context) { go func(cid int64, mid int, stop <-chan struct{}, parentCtx context.Context) {
dots := []string{".", "..", "..."} dots := []string{".", "..", "..."}
@@ -442,9 +455,6 @@ func (c *TelegramChannel) handleMessage(runCtx context.Context, message *telego.
case <-parentCtx.Done(): case <-parentCtx.Done():
return return
case <-stop: case <-stop:
logger.DebugCF("telegram", "Telegram thinking animation stopped", map[string]interface{}{
logger.FieldChatID: cid,
})
return return
case <-ticker.C: case <-ticker.C:
i++ i++
@@ -457,20 +467,11 @@ func (c *TelegramChannel) handleMessage(runCtx context.Context, message *telego.
}) })
cancelEdit() cancelEdit()
if err != nil { if err != nil {
logger.DebugCF("telegram", "Telegram thinking animation edit failed", map[string]interface{}{ return
logger.FieldChatID: cid,
"message_id": mid,
logger.FieldError: err.Error(),
})
} }
} }
} }
}(chatID, pID, stopChan, runCtx) }(chatID, pID, stopChan, runCtx)
} else {
logger.WarnCF("telegram", "Telegram thinking placeholder create failed", map[string]interface{}{
logger.FieldChatID: chatID,
logger.FieldError: err.Error(),
})
} }
metadata := map[string]string{ metadata := map[string]string{
@@ -489,9 +490,6 @@ func (c *TelegramChannel) downloadFile(runCtx context.Context, fileID, ext strin
file, err := c.bot.GetFile(getFileCtx, &telego.GetFileParams{FileID: fileID}) file, err := c.bot.GetFile(getFileCtx, &telego.GetFileParams{FileID: fileID})
cancelGetFile() cancelGetFile()
if err != nil { if err != nil {
logger.WarnCF("telegram", "Failed to get file", map[string]interface{}{
logger.FieldError: err.Error(),
})
return "" return ""
} }
@@ -499,26 +497,12 @@ func (c *TelegramChannel) downloadFile(runCtx context.Context, fileID, ext strin
return "" return ""
} }
// In telego, we can use Link() or just build the URL
url := fmt.Sprintf("https://api.telegram.org/file/bot%s/%s", c.config.Token, file.FilePath) url := fmt.Sprintf("https://api.telegram.org/file/bot%s/%s", c.config.Token, file.FilePath)
logger.DebugCF("telegram", "Telegram file URL resolved", map[string]interface{}{
"url": url,
})
mediaDir := filepath.Join(os.TempDir(), "clawgo_media") mediaDir := filepath.Join(os.TempDir(), "clawgo_media")
if err := os.MkdirAll(mediaDir, 0755); err != nil { _ = os.MkdirAll(mediaDir, 0755)
logger.WarnCF("telegram", "Failed to create media directory", map[string]interface{}{
logger.FieldError: err.Error(),
})
return ""
}
localPath := filepath.Join(mediaDir, fileID[:min(16, len(fileID))]+ext) localPath := filepath.Join(mediaDir, fileID[:min(16, len(fileID))]+ext)
if err := c.downloadFromURL(runCtx, url, localPath); err != nil { if err := c.downloadFromURL(runCtx, url, localPath); err != nil {
logger.WarnCF("telegram", "Failed to download file", map[string]interface{}{
logger.FieldError: err.Error(),
})
return "" return ""
} }
@@ -538,35 +522,28 @@ func (c *TelegramChannel) downloadFromURL(runCtx context.Context, url, localPath
req, err := http.NewRequestWithContext(downloadCtx, http.MethodGet, url, nil) req, err := http.NewRequestWithContext(downloadCtx, http.MethodGet, url, nil)
if err != nil { if err != nil {
return fmt.Errorf("failed to create request: %w", err) return err
} }
client := &http.Client{Timeout: telegramDownloadTimeout} client := &http.Client{Timeout: telegramDownloadTimeout}
resp, err := client.Do(req) resp, err := client.Do(req)
if err != nil { if err != nil {
return fmt.Errorf("failed to download: %w", err) return err
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
return fmt.Errorf("download failed with status: %d", resp.StatusCode) return fmt.Errorf("status: %d", resp.StatusCode)
} }
out, err := os.Create(localPath) out, err := os.Create(localPath)
if err != nil { if err != nil {
return fmt.Errorf("failed to create file: %w", err) return err
} }
defer out.Close() defer out.Close()
_, err = io.Copy(out, resp.Body) _, err = io.Copy(out, resp.Body)
if err != nil { return err
return fmt.Errorf("failed to write file: %w", err)
}
logger.DebugCF("telegram", "File downloaded successfully", map[string]interface{}{
"path": localPath,
})
return nil
} }
func parseChatID(chatIDStr string) (int64, error) { func parseChatID(chatIDStr string) (int64, error) {
@@ -588,23 +565,16 @@ func markdownToTelegramHTML(text string) string {
text = escapeHTML(text) text = escapeHTML(text)
text = regexp.MustCompile(`(?m)^#{1,6}\s+(.+)$`).ReplaceAllString(text, "<b>$1</b>") text = regexp.MustCompile("(?m)^#{1,6}\\s+(.+)$").ReplaceAllString(text, "<b>$1</b>")
text = regexp.MustCompile("(?m)^>\\s*(.*)$").ReplaceAllString(text, "│ $1")
text = regexp.MustCompile(`(?m)^>\s*(.*)$`).ReplaceAllString(text, "│ $1") text = regexp.MustCompile("\\[([^\\]]+)\\]\\(([^)]+)\\)").ReplaceAllString(text, `<a href="$2">$1</a>`)
text = regexp.MustCompile("\\*\\*(.+?)\\*\\*").ReplaceAllString(text, "<b>$1</b>")
text = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`).ReplaceAllString(text, `<a href="$2">$1</a>`) text = regexp.MustCompile("__(.+?)__").ReplaceAllString(text, "<b>$1</b>")
text = regexp.MustCompile("\\*([^*\\n]+)\\*").ReplaceAllString(text, "<i>$1</i>")
text = regexp.MustCompile(`\*\*(.+?)\*\*`).ReplaceAllString(text, "<b>$1</b>") text = regexp.MustCompile("_([^_\\n]+)_").ReplaceAllString(text, "<i>$1</i>")
text = regexp.MustCompile("~~(.+?)~~").ReplaceAllString(text, "<s>$1</s>")
text = regexp.MustCompile(`__(.+?)__`).ReplaceAllString(text, "<b>$1</b>") text = regexp.MustCompile("(?m)^[-*]\\s+").ReplaceAllString(text, "• ")
text = regexp.MustCompile("(?m)^\\d+\\.\\s+").ReplaceAllString(text, "• ")
text = regexp.MustCompile(`\*([^*\n]+)\*`).ReplaceAllString(text, "<i>$1</i>")
text = regexp.MustCompile(`_([^_\n]+)_`).ReplaceAllString(text, "<i>$1</i>")
text = regexp.MustCompile(`~~(.+?)~~`).ReplaceAllString(text, "<s>$1</s>")
text = regexp.MustCompile(`(?m)^[-*]\s+`).ReplaceAllString(text, "• ")
text = regexp.MustCompile(`(?m)^\d+\.\s+`).ReplaceAllString(text, "• ")
for i, code := range inlineCodes.codes { for i, code := range inlineCodes.codes {
escaped := escapeHTML(code) escaped := escapeHTML(code)
@@ -693,8 +663,8 @@ func sanitizeTelegramHTML(input string) string {
return "" return ""
} }
tagRe := regexp.MustCompile(`(?is)<\s*(/?)\s*([a-z0-9]+)([^>]*)>`) tagRe := regexp.MustCompile("(?is)<\\s*(/?)\\s*([a-z0-9]+)([^>]*)>")
hrefRe := regexp.MustCompile(`(?is)\bhref\s*=\s*"([^"]+)"`) hrefRe := regexp.MustCompile("(?is)\\bhref\\s*=\\s*\"([^\"]+)\"")
var out strings.Builder var out strings.Builder
stack := make([]string, 0, 16) stack := make([]string, 0, 16)
@@ -719,7 +689,6 @@ func sanitizeTelegramHTML(input string) string {
} }
if isClose { if isClose {
// Ensure tag stack remains balanced; drop unmatched close tags.
found := -1 found := -1
for i := len(stack) - 1; i >= 0; i-- { for i := len(stack) - 1; i >= 0; i-- {
if stack[i] == tagName { if stack[i] == tagName {
@@ -739,11 +708,9 @@ func sanitizeTelegramHTML(input string) string {
continue continue
} }
// Normalize opening tags; only <a href="..."> can carry attributes.
if tagName == "a" { if tagName == "a" {
hrefMatch := hrefRe.FindStringSubmatch(attrRaw) hrefMatch := hrefRe.FindStringSubmatch(attrRaw)
if len(hrefMatch) < 2 { if len(hrefMatch) < 2 {
// Invalid anchor tag -> degrade to escaped text to avoid parse errors.
out.WriteString("&lt;a&gt;") out.WriteString("&lt;a&gt;")
pos = end pos = end
continue continue
@@ -783,8 +750,7 @@ func escapeHTMLAttr(text string) string {
} }
func plainTextFromTelegramHTML(text string) string { func plainTextFromTelegramHTML(text string) string {
// Best-effort fallback for parse failures: drop tags and keep readable content. tagRe := regexp.MustCompile("(?is)<[^>]+>")
tagRe := regexp.MustCompile(`(?is)<[^>]+>`)
plain := tagRe.ReplaceAllString(text, "") plain := tagRe.ReplaceAllString(text, "")
plain = strings.ReplaceAll(plain, "&lt;", "<") plain = strings.ReplaceAll(plain, "&lt;", "<")
plain = strings.ReplaceAll(plain, "&gt;", ">") plain = strings.ReplaceAll(plain, "&gt;", ">")