organizer/telegram.go

package main

import (
	"context"
	"strings"

	"git.milar.in/milarin/adverr"
	"git.milar.in/milarin/anilist"
	"git.milar.in/milarin/channel"
	"git.milar.in/milarin/slices"
	"git.milar.in/nyaanime/logic"
	"git.milar.in/nyaanime/model"
	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api"
)

var TelegramBot *tgbotapi.BotAPI
var AnimeEpisodeMessageChannel = make(chan model.AnimeEpisode, 1000)

func InitTelegramBot() error {
	if TelegramBotToken != "" && TelegramChatID != 0 {
		bot, err := tgbotapi.NewBotAPI(TelegramBotToken)
		if err != nil {
			return err
		}
		TelegramBot = bot
	}

	// close channel on app shutdown
	go func() {
		<-AppCtx.Done()
		close(AnimeEpisodeMessageChannel)
	}()

	AppExitWg.Add(1)
	go SendMessagePeriodically()

	return nil
}

func SendMessagePeriodically() {
	defer AppExitWg.Done()

	var messagesPerInterval <-chan []model.AnimeEpisode

	if TelegramOrganizeMessageSendInterval > 0 {
		WaitForNextTelegramSendCycle()
		sendAllQueuedAnimeEpisodes()

		grouperFunc := func(current []model.AnimeEpisode, value model.AnimeEpisode) []model.AnimeEpisode {
			return append(current, value)
		}
		messagesPerInterval = channel.GroupByTime(AnimeEpisodeMessageChannel, TelegramOrganizeMessageSendInterval, grouperFunc)
	} else {
		mapperFunc := func(value model.AnimeEpisode) []model.AnimeEpisode {
			return []model.AnimeEpisode{value}
		}
		messagesPerInterval = channel.MapSuccessive(AnimeEpisodeMessageChannel, mapperFunc)
	}

	for animeEpisodes := range messagesPerInterval {
		sendTelegramAnimeEpMessage(animeEpisodes)
	}
}

func sendAllQueuedAnimeEpisodes() {
	queuedEpisodeAmount := len(AnimeEpisodeMessageChannel)
	animeEpisodes := make([]model.AnimeEpisode, 0, queuedEpisodeAmount)
	for i := 0; i < queuedEpisodeAmount; i++ {
		animeEpisodes = append(animeEpisodes, <-AnimeEpisodeMessageChannel)
	}
	sendTelegramAnimeEpMessage(animeEpisodes)
}

func SendTelegramMessage(text string) {
	if TelegramBot == nil || strings.TrimSpace(text) == "" {
		return
	}

	msg := tgbotapi.NewMessage(TelegramChatID, text)
	msg.ParseMode = "html"
	_, err := TelegramBot.Send(msg)
	if err != nil {
		adverr.Println(adverr.Wrap("could not send telegram message", err))
	}
}

func PrepareTelegramAnimeEpMessage(animeEp model.AnimeEpisode) {
	shouldSendMessage, err := CheckSendConditions(animeEp)
	if err != nil {
		adverr.Println(adverr.Wrap("could not check telegram message send conditions", err))
	}

	if !shouldSendMessage {
		return
	}

	AnimeEpisodeMessageChannel <- animeEp
}

func sendTelegramAnimeEpMessage(animeEpisodes []model.AnimeEpisode) {
	if len(animeEpisodes) == 0 {
		return
	}

	animeEpisodes = RemoveDuplicates(animeEpisodes)

	b := new(strings.Builder)
	if err := TelegramOrganizeMessagePattern.Execute(b, animeEpisodes); err != nil {
		adverr.Println(adverr.Wrap("could not send telegram message", err))
	}

	SendTelegramMessage(b.String())
}

func RemoveDuplicates(animeEpisodes []model.AnimeEpisode) []model.AnimeEpisode {
	mapperFunc := func(animeEp model.AnimeEpisode) (model.Pair[anilist.MediaID, int], model.AnimeEpisode) {
		return model.Pair[anilist.MediaID, int]{First: animeEp.Anime.ID, Second: animeEp.Episode}, animeEp
	}

	unmapperFunc := func(key model.Pair[anilist.MediaID, int], value model.AnimeEpisode) model.AnimeEpisode {
		return value
	}

	return slices.OfMap(slices.ToMap(animeEpisodes, mapperFunc), unmapperFunc)
}

func CheckSendConditions(animeEp model.AnimeEpisode) (bool, error) {
	listEntry, err := GetListEntry(animeEp.Anime)
	if err != nil {
		return false, err
	}

	for _, sendCondition := range AllSendConditions {
		// check if user configured current sendCondition
		if !slices.Contains(TelegramOrganizeMessageSendCondition, sendCondition) {
			continue
		}

		// check if current sendCondition applies for given anime episode
		if sendCondition.ShouldSend(animeEp, listEntry) {
			return true, nil
		}
	}

	return false, nil
}

func GetListEntry(anime *anilist.Media) (*anilist.MediaList, error) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	listEntries, err := logic.GetCurrentlyWatchingAnimesContext(ctx, logic.AllMediaListStatuses...)
	if err != nil {
		return nil, err
	}

	filteredListEntries := channel.Filter(listEntries, func(a *anilist.MediaList) bool { return a.MediaID == anime.ID })
	listEntry := channel.FindFirstAndCancelFlush(filteredListEntries, cancel) // TODO flush working properly?

	if listEntry == nil {
		return nil, nil
	}

	return *listEntry, nil
}