342 lines
11 KiB
Go
342 lines
11 KiB
Go
package workers
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.savin.nyc/alex/go-receipt-tracker/bus"
|
|
"git.savin.nyc/alex/go-receipt-tracker/config"
|
|
"git.savin.nyc/alex/go-receipt-tracker/models"
|
|
tele "gopkg.in/telebot.v4"
|
|
)
|
|
|
|
type Bot struct {
|
|
id string //
|
|
opts *config.Telegram //
|
|
client *tele.Bot //
|
|
subscriptions map[string]chan bus.Event //
|
|
bus *bus.Bus //
|
|
log *slog.Logger //
|
|
cancel context.CancelFunc //
|
|
end uint32 // ensure the close methods are only called once
|
|
}
|
|
|
|
func NewTelegramBot(cfg *config.Telegram, b *bus.Bus) *Bot {
|
|
|
|
bot := &Bot{
|
|
id: "telegram",
|
|
opts: cfg,
|
|
subscriptions: make(map[string]chan bus.Event),
|
|
bus: b,
|
|
}
|
|
|
|
return bot
|
|
}
|
|
|
|
func (b *Bot) ID() string {
|
|
return b.id
|
|
}
|
|
|
|
func (b *Bot) Type() string {
|
|
return "worker"
|
|
}
|
|
|
|
func (b *Bot) Init(log *slog.Logger) error {
|
|
var err error
|
|
b.log = log
|
|
|
|
opts := tele.Settings{
|
|
Token: b.opts.ApiKey,
|
|
Poller: &tele.LongPoller{
|
|
Timeout: 10 * time.Second,
|
|
},
|
|
}
|
|
|
|
b.client, err = tele.NewBot(opts)
|
|
if err != nil {
|
|
b.log.Debug("cannot create telegram bot", "error", err)
|
|
return err
|
|
}
|
|
|
|
b.client.Handle(tele.OnText, func(c tele.Context) error {
|
|
b.log.Debug("received a new text message", "chat", c.Chat().ID, "message", c.Message().Text)
|
|
|
|
b.client.Notify(c.Chat(), tele.ChatAction("typing"))
|
|
|
|
err = b.bus.Publish("processor:text", c.Message().Text)
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to a channel", "channel", "processor:text", "error", err.Error())
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
b.client.Handle(tele.OnPhoto, func(c tele.Context) error {
|
|
b.client.Notify(c.Chat(), tele.ChatAction("typing"))
|
|
|
|
photo := c.Message().Photo
|
|
b.client.Download(&photo.File, b.opts.StoragePath+"/"+photo.UniqueID+".jpg")
|
|
|
|
msg := &bus.Message{
|
|
Image: &bus.Image{
|
|
ID: photo.UniqueID,
|
|
Filename: b.opts.StoragePath + "/" + photo.UniqueID + ".jpg",
|
|
},
|
|
ReplyType: "send",
|
|
TbContext: c,
|
|
}
|
|
err := b.bus.Publish("processor:image", msg)
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to the channel", "channel", "processor:image", "error", err.Error())
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
b.client.Handle(tele.OnCallback, func(c tele.Context) error {
|
|
tbc := new(models.TelegramBotCommand)
|
|
err := tbc.Unmarshal(c.Callback().Data)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
switch tbc.Cmd {
|
|
case models.BtnCmd["user_add"]:
|
|
return nil
|
|
case models.BtnCmd["user_edit"]:
|
|
msg := &bus.Message{
|
|
TBCmd: tbc,
|
|
TbContext: c,
|
|
ReplyType: "send",
|
|
}
|
|
err := b.bus.Publish("processor:user_edit", msg)
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to the channel", "channel", "processor:user_edit", "error", err.Error())
|
|
}
|
|
|
|
return nil
|
|
case models.BtnCmd["user_delete"]:
|
|
// b.client.Notify(c.Chat(), tele.ChatAction("typing"))
|
|
// msg.Text = "Welcome back, " + event.Payload.(*bus.Message).TbContext.Sender().FirstName + " " + event.Payload.(*bus.Message).TbContext.Sender().LastName
|
|
// msg.InlineKeyboard = &tele.ReplyMarkup{
|
|
// InlineKeyboard: [][]tele.InlineButton{
|
|
// {
|
|
// {Text: "Yes, Delete", Data: models.String(&models.TelegramBotCommand{Cmd: models.BtnCmd["user_delete"], ID: msg.TbContext.Sender().ID})},
|
|
// {Text: "No, Cancel", Data: models.String(&models.TelegramBotCommand{Cmd: models.BtnCmd["user_delete"], ID: msg.TbContext.Sender().ID})},
|
|
// },
|
|
// {
|
|
// {Text: "« Back to Receipt", Data: models.String(&models.TelegramBotCommand{Cmd: models.BtnCmd["user_edit"], ID: msg.TbContext.Sender().ID})},
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
return nil
|
|
case models.BtnCmd["receipt_add"]:
|
|
return nil
|
|
case models.BtnCmd["receipt_edit"]:
|
|
return nil
|
|
case models.BtnCmd["receipts_list"]:
|
|
return nil
|
|
case models.BtnCmd["items_list"]:
|
|
cbr := &tele.CallbackResponse{
|
|
CallbackID: c.Callback().ID,
|
|
Text: "You clicked the button!",
|
|
ShowAlert: false,
|
|
}
|
|
c.Respond(cbr)
|
|
msg := &bus.Message{
|
|
TBCmd: tbc,
|
|
TbContext: c,
|
|
ReplyType: "send",
|
|
}
|
|
err := b.bus.Publish("processor:items_list", msg)
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to the channel", "channel", "processor:items_list", "error", err.Error())
|
|
}
|
|
// r := new(Receipt)
|
|
// err = db.NewSelect().Model(r).Where("receipt_id = ?", tbc.ID).Scan(ctx)
|
|
// if err != nil {
|
|
// panic(err)
|
|
// }
|
|
|
|
// var items []Item
|
|
// _, err := db.NewSelect().Model(&items).Where("receipt_id = ?", tbc.ID).Limit(10).ScanAndCount(ctx)
|
|
// if err != nil {
|
|
// panic(err)
|
|
// }
|
|
|
|
// message := "<code>List of the Receipt Items:\n" + strings.Repeat("-", 32) + "\n"
|
|
|
|
// merchant := truncateText(r.Merchant, 16)
|
|
// spaces := 32 - (len("Merchant:") + len(merchant))
|
|
// message += "Merchant:" + strings.Repeat(" ", spaces) + merchant + "\n"
|
|
// spaces = 32 - (len("Date/Time:") + len(r.PaidAt))
|
|
// message += "Date/Time:" + strings.Repeat(" ", spaces) + r.PaidAt + "\n" + strings.Repeat("-", 32) + "\n"
|
|
|
|
// keyboard := &tele.ReplyMarkup{InlineKeyboard: [][]tele.InlineButton{}}
|
|
|
|
// for _, item := range items {
|
|
// // b := tele.InlineButton{Text: fmt.Sprintf("%s (%.2f) $%.2f", strings.Title(strings.ToLower(item.Title)), item.Quantity, item.Price), Data: String(&TelegramBotCommand{Cmd: btnCmd["item_edit"], ID: item.ItemID})}
|
|
// // keyboard.InlineKeyboard = append(keyboard.InlineKeyboard, []tele.InlineButton{b})
|
|
// // if item.Quantity == math.Trunc(item.Quantity) {
|
|
// // fmt.Sprintf("%d", math.Trunc(item.Quantity))
|
|
// // } else {
|
|
// // fmt.Sprintf("%.2f", item.Quantity)
|
|
// // }
|
|
// title := truncateText(item.Title, 16)
|
|
// spaces := 32 - (len(title) + len("x"+fmt.Sprintf("%+v", item.Quantity)) + len(fmt.Sprintf("$%.2f", item.Price)) + 3)
|
|
// message += title + strings.Repeat(".", spaces) + "x" + fmt.Sprintf("%+v", item.Quantity) + "..." + fmt.Sprintf("$%.2f", item.Price) + "\n"
|
|
// log.Printf("ITEM: %s // %.2f // %.2f", strings.Title(strings.ToLower(item.Title)), item.Quantity, item.Price)
|
|
// }
|
|
// // keyboard.InlineKeyboard = append(keyboard.InlineKeyboard, []tele.InlineButton{{Text: "«", Data: String(&TelegramBotCommand{Cmd: btnPage["prev"], ID: tbc.ID})}, {Text: "»", Data: String(&TelegramBotCommand{Cmd: btnPage["next"], ID: tbc.ID})}})
|
|
// keyboard.InlineKeyboard = append(keyboard.InlineKeyboard, []tele.InlineButton{{Text: "« Back to Receipt", Data: String(&TelegramBotCommand{Cmd: btnCmd["receipt_edit"], ID: tbc.ID})}})
|
|
|
|
// message += strings.Repeat("-", 32) + "\n"
|
|
// spaces = 32 - (len("Tax:") + len(fmt.Sprintf("$%.2f", r.Tax)))
|
|
// message += "Tax:" + strings.Repeat(" ", spaces) + fmt.Sprintf("$%.2f", r.Tax) + "\n"
|
|
// if r.CCFee > 0 {
|
|
// spaces = 32 - (len("CC Fee:") + len(fmt.Sprintf("$%.2f", r.CCFee)))
|
|
// message += "CC Fee:" + strings.Repeat(" ", spaces) + fmt.Sprintf("$%.2f", r.CCFee) + "\n"
|
|
// }
|
|
// spaces = 32 - (len("Total:") + len(fmt.Sprintf("$%.2f", r.Total)))
|
|
// message += "Total:" + strings.Repeat(" ", spaces) + fmt.Sprintf("<b>$%.2f</b>", r.Total) + "\n</code>"
|
|
|
|
// opts := &tele.SendOptions{ParseMode: "HTML", ReplyMarkup: keyboard}
|
|
|
|
// return c.EditCaption(message, opts)
|
|
return nil
|
|
}
|
|
// cbr := &tele.CallbackResponse{
|
|
// CallbackID: c.Callback().ID,
|
|
// Text: "No unique",
|
|
// ShowAlert: false,
|
|
// }
|
|
// return c.Respond(cbr)
|
|
return nil
|
|
})
|
|
|
|
b.client.Handle("/start", func(c tele.Context) error {
|
|
b.client.Notify(c.Chat(), tele.ChatAction("typing"))
|
|
msg := &bus.Message{
|
|
ReplyType: "send",
|
|
TbContext: c,
|
|
}
|
|
|
|
err = b.bus.Publish("processor:user_add", msg)
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to a channel", "channel", "processor:user_add", "error", err.Error())
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
b.client.Handle("/receipts", func(c tele.Context) error {
|
|
b.log.Debug("publish message to a channel", "channel", "processor:receipts_list", "message", "Test Message")
|
|
b.client.Notify(c.Chat(), tele.ChatAction("typing"))
|
|
|
|
err = b.bus.Publish("processor:receipts_list", "Test Message")
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to a channel", "channel", "processor:receipts_list", "error", err.Error())
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
b.client.Handle("/reports", func(c tele.Context) error {
|
|
b.log.Debug("publish message to a channel", "channel", "processor:reports_list", "message", "Test Message")
|
|
b.client.Notify(c.Chat(), tele.ChatAction("typing"))
|
|
|
|
err = b.bus.Publish("processor:reports_list", "Test Message")
|
|
if err != nil {
|
|
b.log.Error("couldn't publish to a channel", "channel", "processor:reports_list", "error", err.Error())
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
func (b *Bot) OneTime() error {
|
|
return nil
|
|
}
|
|
|
|
func (b *Bot) Serve() {
|
|
if atomic.LoadUint32(&b.end) == 1 {
|
|
return
|
|
}
|
|
|
|
b.subscribe("telegram:send")
|
|
b.subscribe("telegram:reply")
|
|
b.subscribe("telegram:edit_caption")
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
b.cancel = cancel
|
|
|
|
go b.eventLoop(ctx)
|
|
|
|
time.Sleep(3 * time.Second)
|
|
|
|
go b.client.Start()
|
|
|
|
// if atomic.LoadUint32(&b.end) == 0 {
|
|
// go func() {
|
|
// if !w.mqtt.IsConnected() {
|
|
// b.bus.Publish("app:connected", &bus.ConnectionStatus{WorkerID: b.ID(), WorkerType: b.Type(), IsConnected: false})
|
|
// b.log.Warn(b.ID() + " client is disconnected")
|
|
// }
|
|
// }()
|
|
// }
|
|
}
|
|
|
|
func (b *Bot) Stop() {
|
|
b.client.Stop()
|
|
}
|
|
|
|
func (b *Bot) Close() {
|
|
b.cancel()
|
|
}
|
|
|
|
func (b *Bot) subscribe(chn string) error {
|
|
s, err := b.bus.Subscribe(chn, b.Type()+":"+b.ID())
|
|
if err != nil {
|
|
b.log.Error("couldn't subscribe to a channel", "channel", chn, "error", err.Error())
|
|
return err
|
|
}
|
|
b.subscriptions[chn] = s
|
|
|
|
return nil
|
|
}
|
|
|
|
// eventLoop loops forever
|
|
func (b *Bot) eventLoop(ctx context.Context) {
|
|
b.log.Debug(b.ID() + " communication event loop started")
|
|
defer b.log.Debug(b.ID() + " communication event loop halted")
|
|
|
|
for {
|
|
select {
|
|
|
|
case event := <-b.subscriptions["telegram:send"]:
|
|
b.log.Debug("worker got a new message from a channel", "channel", event.ChannelName, "type", b.Type(), "id", b.ID())
|
|
msg := event.Payload.(*bus.Message)
|
|
b.client.Send(msg.TbContext.Sender(), msg.Text, msg.InlineKeyboard)
|
|
|
|
case event := <-b.subscriptions["telegram:reply"]:
|
|
b.log.Debug("worker got a new message from a channel", "channel", event.ChannelName, "type", b.Type(), "id", b.ID())
|
|
msg := event.Payload.(*bus.Message)
|
|
b.client.Reply(msg.TbContext.Message(), msg.Text, msg.InlineKeyboard)
|
|
|
|
case event := <-b.subscriptions["telegram:edit_caption"]:
|
|
b.log.Debug("worker got a new message from a channel", "channel", event.ChannelName, "type", b.Type(), "id", b.ID())
|
|
msg := event.Payload.(*bus.Message)
|
|
b.client.EditCaption(msg.TbContext.Message(), msg.Text, msg.InlineKeyboard)
|
|
|
|
case <-ctx.Done():
|
|
b.log.Info("stopping " + b.ID() + " communication event loop")
|
|
return
|
|
}
|
|
}
|
|
}
|