Diffstat (limited to 'internal/transcoder/telegram.go')
| -rw-r--r-- | internal/transcoder/telegram.go | 584 |
1 files changed, 0 insertions, 584 deletions
diff --git a/internal/transcoder/telegram.go b/internal/transcoder/telegram.go deleted file mode 100644 index b521fb0..0000000 --- a/internal/transcoder/telegram.go +++ /dev/null @@ -1,584 +0,0 @@ -package transcoder - -import ( - "bytes" - "encoding/json" - "fmt" - "io" - "log/slog" - "mime/multipart" - "net/http" - "os" - "path/filepath" - "slices" - "strconv" - "strings" - "time" - - "github.com/SayaAndy/saya-today-article-metadata-add/config" - "github.com/SayaAndy/saya-today-article-metadata-add/internal/draft" -) - -const ( - telegramMediaGroupLimit = 10 - // telegramCaptionLimit is the max length (runes) of a media caption. - telegramCaptionLimit = 1024 - // telegramMaxAttempts caps retries when the API returns 429. - telegramMaxAttempts = 5 -) - -type TelegramTranscoder struct { - botToken string - channelID int - groupID int - lastUpdateID int - imageBaseURL string - previewBaseURL string - previewExtension string - stateDir string - http *http.Client -} - -func NewTelegramTranscoder(cfg any) (Transcoder, error) { - tgCfg, ok := cfg.(*config.TelegramConfig) - if !ok { - return nil, fmt.Errorf("invalid config for telegram transcoder") - } - - transcoder := &TelegramTranscoder{ - botToken: tgCfg.BotToken, - channelID: tgCfg.ChannelID, - groupID: tgCfg.GroupID, - lastUpdateID: -1, - imageBaseURL: strings.TrimRight(tgCfg.ImageBaseURL, "/"), - previewBaseURL: strings.TrimRight(tgCfg.PreviewBaseURL, "/"), - previewExtension: tgCfg.PreviewExtension, - stateDir: tgCfg.StateDir, - http: &http.Client{Timeout: time.Minute}, - } - - if _, err := transcoder.getUpdates(); err != nil { - return nil, fmt.Errorf("failed to get last update id for checkpoint: %w", err) - } - - return transcoder, nil -} - -func (t *TelegramTranscoder) Name() string { return "telegram" } - -func (t *TelegramTranscoder) Finalize() error { return nil } - -// telegramState records, per draft, which section indices have already been -// posted and the message ids they produced. Posting is append-only: a section -// already present here is never re-sent. -type telegramState struct { - ChannelID string `json:"channelID"` - Posted map[string][]int `json:"posted"` -} - -func (t *TelegramTranscoder) Transcode(doc *draft.Document) error { - statePath := filepath.Join(t.stateDir, doc.Codename+".telegram-state.json") - state, err := loadTelegramState(statePath) - if err != nil { - return fmt.Errorf("load telegram state %q: %w", statePath, err) - } - - if state.ChannelID != "" && state.ChannelID != strconv.Itoa(t.channelID) { - slog.Warn("telegram channel changed, re-posting all sections to the new channel", - slog.String("codename", doc.Codename), - slog.String("old", state.ChannelID), - slog.Int("new", t.channelID)) - state.Posted = map[string][]int{} - } - state.ChannelID = strconv.Itoa(t.channelID) - - for _, section := range doc.Sections { - idx := strconv.Itoa(section.Index) - if _, done := state.Posted[idx]; done { - continue - } - - var msgIDs []int - pendingText := "" - - // flushText sends any buffered text as its own message. - flushText := func() error { - if pendingText == "" { - return nil - } - id, err := t.sendMessage(pendingText) - if err != nil { - return err - } - msgIDs = append(msgIDs, id) - pendingText = "" - return nil - } - - for _, block := range section.Blocks { - switch block.Kind { - case draft.TextBlock: - text := stripRules(block.Text) - if text == "" { - continue - } - // flush any earlier text before buffering this one - if err := flushText(); err != nil { - return fmt.Errorf("send section %d text: %w", section.Index, err) - } - pendingText = text - case draft.GalleryBlock: - // Attach the immediately preceding text as the album caption when - // it fits; otherwise send it as its own message first. - caption := "" - if pendingText != "" { - if len([]rune(pendingText)) <= telegramCaptionLimit { - caption = pendingText - pendingText = "" - } else if err := flushText(); err != nil { - return fmt.Errorf("send section %d text: %w", section.Index, err) - } - } - for i, group := range splitPhotos(block.Gallery.Photos, telegramMediaGroupLimit) { - groupCaption := "" - if i == 0 { - groupCaption = caption - } - ids, err := t.sendMediaGroup(group, groupCaption, "photo", t.channelID, 0) - if err != nil { - return fmt.Errorf("send section %d gallery: %w", section.Index, err) - } - msgIDs = append(msgIDs, ids...) - if err := flushText(); err != nil { - return fmt.Errorf("send section %d text: %w", section.Index, err) - } - if len(msgIDs) > 0 { - time.Sleep(5 * time.Second) - forwardedID, err := t.getForwardedID(msgIDs) - if err != nil { - return fmt.Errorf("send full documents for section %d gallery: %w", section.Index, err) - } - if forwardedID == 0 { - return fmt.Errorf("fail to get forwarded id of section %d gallery", section.Index) - } - _, err = t.sendMediaGroup(group, "", "document", t.groupID, forwardedID) - if err != nil { - return fmt.Errorf("send section %d documents: %w", section.Index, err) - } - } - } - } - } - - state.Posted[idx] = msgIDs - if err := saveTelegramState(statePath, state); err != nil { - return fmt.Errorf("save telegram state %q: %w", statePath, err) - } - slog.Info("telegram posted section", - slog.String("codename", doc.Codename), - slog.Int("section", section.Index), - slog.Int("messages", len(msgIDs))) - } - - return nil -} - -func loadTelegramState(path string) (*telegramState, error) { - raw, err := os.ReadFile(path) - if err != nil { - if os.IsNotExist(err) { - return &telegramState{Posted: map[string][]int{}}, nil - } - return nil, err - } - var state telegramState - if err := json.Unmarshal(raw, &state); err != nil { - return nil, err - } - if state.Posted == nil { - state.Posted = map[string][]int{} - } - return &state, nil -} - -func saveTelegramState(path string, state *telegramState) error { - if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { - return err - } - raw, err := json.MarshalIndent(state, "", " ") - if err != nil { - return err - } - return os.WriteFile(path, raw, 0o644) -} - -// stripRules drops standalone "---" horizontal-rule lines (meaningless in a -// Telegram message) and trims the remaining text. -func stripRules(text string) string { - lines := strings.Split(text, "\n") - kept := lines[:0] - for _, line := range lines { - if strings.TrimSpace(line) == "---" { - continue - } - kept = append(kept, line) - } - return strings.TrimSpace(strings.Join(kept, "\n")) -} - -// splitPhotos divides photos into near-equal groups each no larger than limit. -func splitPhotos(photos []draft.Photo, limit int) [][]draft.Photo { - n := len(photos) - if n == 0 { - return nil - } - groupCount := (n + limit - 1) / limit - base := n / groupCount - rem := n % groupCount - - var groups [][]draft.Photo - start := 0 - for g := range groupCount { - size := base - if g < rem { - size++ - } - groups = append(groups, photos[start:start+size]) - start += size - } - return groups -} - -type inputMediaPhoto struct { - Type string `json:"type"` - Media string `json:"media"` - Caption string `json:"caption,omitempty"` - ParseMode string `json:"parse_mode,omitempty"` -} - -func (t *TelegramTranscoder) photoURL(key string) string { - return t.imageBaseURL + "/" + key -} - -func (t *TelegramTranscoder) previewURL(key string) string { - base := t.previewBaseURL - if base == "" { - base = t.imageBaseURL - } - if t.previewExtension != "" { - key = strings.TrimSuffix(key, filepath.Ext(key)) + t.previewExtension - } - return base + "/" + key -} - -func (t *TelegramTranscoder) sendMessage(text string) (int, error) { - payload := map[string]any{ - "chat_id": strconv.Itoa(t.channelID), - "text": markdownToTelegramHTML(text), - "parse_mode": "HTML", - "disable_web_page_preview": true, - } - var result struct { - MessageID int `json:"message_id"` - } - if err := t.call("sendMessage", payload, &result); err != nil { - return 0, err - } - return result.MessageID, nil -} - -func (t *TelegramTranscoder) sendMediaGroup(photos []draft.Photo, caption string, sendAsType string, id int, replyTo int) ([]int, error) { - media := make([]inputMediaPhoto, 0, len(photos)) - files := map[string]uploadFile{} - for i, p := range photos { - item := inputMediaPhoto{Type: sendAsType} - if sendAsType == "document" { - // Upload the original file directly rather than passing Telegram a - // URL to fetch: full-resolution originals routinely fail Telegram's - // server-side downloader with WEBPAGE_CURL_FAILED. - data, err := t.downloadPhoto(p.Key) - if err != nil { - return nil, fmt.Errorf("download %q for upload: %w", p.Key, err) - } - field := fmt.Sprintf("file%d", i) - files[field] = uploadFile{name: filepath.Base(p.Key), data: data} - item.Media = "attach://" + field - } else { - item.Media = t.previewURL(p.Key) - } - // Telegram shows an album's caption in the feed only when exactly one - // item is captioned. So the section text goes on the first photo and - // every other photo is left uncaptioned (per-photo captions dropped). - if i == 0 && caption != "" { - item.Caption = markdownToTelegramHTML(caption) - item.ParseMode = "HTML" - } - media = append(media, item) - } - - var result []struct { - MessageID int `json:"message_id"` - } - - if len(files) > 0 { - mediaJSON, err := json.Marshal(media) - if err != nil { - return nil, err - } - fields := map[string]string{ - "chat_id": strconv.Itoa(id), - "media": string(mediaJSON), - } - if replyTo != 0 { - replyJSON, err := json.Marshal(map[string]any{"message_id": replyTo}) - if err != nil { - return nil, err - } - fields["reply_parameters"] = string(replyJSON) - } - if err := t.callMultipart("sendMediaGroup", fields, files, &result); err != nil { - return nil, err - } - } else { - payload := map[string]any{ - "chat_id": strconv.Itoa(id), - "media": media, - } - if replyTo != 0 { - payload["reply_parameters"] = map[string]any{ - "message_id": replyTo, - } - } - if err := t.call("sendMediaGroup", payload, &result); err != nil { - return nil, err - } - } - - ids := make([]int, 0, len(result)) - for _, r := range result { - ids = append(ids, r.MessageID) - } - return ids, nil -} - -type uploadFile struct { - name string - data []byte -} - -// downloadPhoto fetches the original bytes for a photo key so they can be -// uploaded directly to Telegram instead of fetched server-side by Telegram. -func (t *TelegramTranscoder) downloadPhoto(key string) ([]byte, error) { - url := t.photoURL(key) - resp, err := t.http.Get(url) - if err != nil { - return nil, err - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status %d fetching %q", resp.StatusCode, url) - } - return io.ReadAll(resp.Body) -} - -func (t *TelegramTranscoder) getForwardedID(channelMessageIDs []int) (int, error) { - var payloads []getUpdatesPayload - var err error - slog.Debug("search for forwarded message id", slog.Any("original_id", channelMessageIDs), slog.Int("group_id", t.groupID), slog.Int("channel_id", t.channelID)) - - for range 3 { - for payloads, err = t.getUpdates(); len(payloads) > 0 && err == nil; payloads, err = t.getUpdates() { - for _, payload := range payloads { - if payload.Message.MessageOrigin.Type == "channel" && - payload.Message.MessageOrigin.Chat.ID == t.channelID && - slices.Contains(channelMessageIDs, payload.Message.MessageOrigin.MessageID) && - payload.Message.Chat.ID == t.groupID { - return payload.Message.MessageID, nil - } - } - } - time.Sleep(3 * time.Second) - } - - return 0, err -} - -type getUpdatesPayload struct { - UpdateID int `json:"update_id"` - Message struct { - MessageID int `json:"message_id"` - Chat struct { - ID int `json:"id"` - Title string `json:"title"` - Type string `json:"type"` - } `json:"chat"` - MessageOrigin struct { - Type string `json:"type"` - Date int `json:"date"` - Chat struct { - ID int `json:"id"` - Title string `json:"title"` - Type string `json:"type"` - } `json:"chat"` - MessageID int `json:"message_id"` - } `json:"forward_origin"` - Date int `json:"date"` - Text string `json:"text"` - } `json:"message"` -} - -func (t *TelegramTranscoder) getUpdates() ([]getUpdatesPayload, error) { - var out []getUpdatesPayload - - payload := struct { - Offset int `json:"offset"` - }{ - Offset: t.lastUpdateID, - } - - if err := t.call("getUpdates", payload, &out); err != nil { - return nil, fmt.Errorf("failed to transcode received message: %w", err) - } - - for _, p := range out { - pjson, _ := json.Marshal(p) - slog.Debug("parsed new update", slog.String("payload", string(pjson))) - if p.UpdateID > t.lastUpdateID { - t.lastUpdateID = p.UpdateID - } - } - t.lastUpdateID++ - - return out, nil -} - -// handleResponse decodes a Telegram API response envelope. When the API asks us -// to back off (HTTP 429) it returns a positive wait duration and a nil error so -// the caller can retry; otherwise it unmarshals the result into out (if any) or -// returns the API error verbatim. -func (t *TelegramTranscoder) handleResponse(method string, resp *http.Response, attempt int, out any) (time.Duration, error) { - var envelope struct { - OK bool `json:"ok"` - ErrorCode int `json:"error_code"` - Description string `json:"description"` - Parameters struct { - RetryAfter int `json:"retry_after"` - } `json:"parameters"` - Result json.RawMessage `json:"result"` - } - decodeErr := json.NewDecoder(resp.Body).Decode(&envelope) - resp.Body.Close() - if decodeErr != nil { - return 0, fmt.Errorf("decode telegram response (%s): %w", method, decodeErr) - } - - envelopeBytes, _ := json.Marshal(envelope) - slog.Debug("telegram response", slog.String("method", method), slog.String("response", string(envelopeBytes))) - - if envelope.OK { - if out == nil { - return 0, nil - } - return 0, json.Unmarshal(envelope.Result, out) - } - - // 429 Too Many Requests: Telegram tells us how long to wait in - // parameters.retry_after. Back off and retry instead of failing. - if envelope.ErrorCode == 429 && attempt < telegramMaxAttempts { - wait := time.Duration(envelope.Parameters.RetryAfter) * time.Second - if wait <= 0 { - wait = time.Second - } - slog.Warn("telegram rate limited, backing off", - slog.String("method", method), - slog.Int("attempt", attempt), - slog.Duration("retry_after", wait)) - return wait, nil - } - - return 0, fmt.Errorf("telegram %s failed: %s", method, envelope.Description) -} - -// call invokes a Telegram Bot API method with a JSON body and decodes the -// "result" field into out. The API's error description is surfaced verbatim. -func (t *TelegramTranscoder) call(method string, payload any, out any) error { - body, err := json.Marshal(payload) - if err != nil { - return err - } - - url := fmt.Sprintf("https://api.telegram.org/bot%s/%s", t.botToken, method) - - for attempt := 1; ; attempt++ { - req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/json") - - slog.Debug("send message", slog.String("method", method), slog.String("request", string(body))) - - resp, err := t.http.Do(req) - if err != nil { - return err - } - - wait, err := t.handleResponse(method, resp, attempt, out) - if wait > 0 { - time.Sleep(wait) - continue - } - return err - } -} - -// callMultipart invokes a Telegram Bot API method using multipart/form-data, -// uploading the given files directly instead of handing Telegram URLs to fetch -// server-side. fields carries the non-file form values (chat_id, media JSON, -// reply_parameters, ...); files maps a form field name to its bytes and upload -// filename, referenced from the media JSON via attach://<field>. -func (t *TelegramTranscoder) callMultipart(method string, fields map[string]string, files map[string]uploadFile, out any) error { - url := fmt.Sprintf("https://api.telegram.org/bot%s/%s", t.botToken, method) - - for attempt := 1; ; attempt++ { - var body bytes.Buffer - w := multipart.NewWriter(&body) - for k, v := range fields { - if err := w.WriteField(k, v); err != nil { - return err - } - } - for field, f := range files { - part, err := w.CreateFormFile(field, f.name) - if err != nil { - return err - } - if _, err := part.Write(f.data); err != nil { - return err - } - } - if err := w.Close(); err != nil { - return err - } - - req, err := http.NewRequest(http.MethodPost, url, &body) - if err != nil { - return err - } - req.Header.Set("Content-Type", w.FormDataContentType()) - - slog.Debug("send media group upload", slog.String("method", method), slog.Int("files", len(files))) - - resp, err := t.http.Do(req) - if err != nil { - return err - } - - wait, err := t.handleResponse(method, resp, attempt, out) - if wait > 0 { - time.Sleep(wait) - continue - } - return err - } -} |