feat: repurpose metadata parser as article transcoder for saya.uz andmain

telegram

1 files changed, 584 insertions, 0 deletions

diff --git a/internal/transcoder/telegram.go b/internal/transcoder/telegram.go
new file mode 100644
index 0000000..b521fb0
--- /dev/null
+++ b/internal/transcoder/telegram.go
@@ -0,0 +1,584 @@
+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
+	}
+}