feat(upload): add support on upload to topic

app/up/elem.go

@@ -11,9 +11,10 @@ import (
 )
 
 type iterElem struct {
-	file  *uploaderFile
-	thumb *uploaderFile
-	to    peers.Peer
+	file   *uploaderFile
+	thumb  *uploaderFile
+	to     peers.Peer
+	thread int
 
 	asPhoto bool
 	remove  bool
@@ -34,6 +35,10 @@ func (e *iterElem) To() tg.InputPeerClass {
 	return e.to.InputPeer()
 }
 
+func (e *iterElem) Thread() int {
+	return e.thread
+}
+
 func (e *iterElem) AsPhoto() bool {
 	return e.asPhoto
 }

app/up/iter.go

@@ -2,38 +2,84 @@ package up
 
 import (
 	"context"
+	"github.com/expr-lang/expr/vm"
+	"github.com/gotd/td/telegram/peers"
+	"github.com/iyear/tdl/pkg/logger"
+	"github.com/iyear/tdl/pkg/texpr"
+	"github.com/mitchellh/mapstructure"
+	"go.uber.org/zap"
 	"os"
+	"path/filepath"
+	"strings"
 
 	"github.com/gabriel-vasile/mimetype"
 	"github.com/go-faster/errors"
-	"github.com/gotd/td/telegram/peers"
-
 	"github.com/iyear/tdl/pkg/uploader"
 	"github.com/iyear/tdl/pkg/utils"
 )
 
-type file struct {
-	file  string
-	thumb string
+type Env struct {
+	File      string `comment:"File path"`
+	Thumb     string `comment:"Thumbnail path"`
+	Filename  string `comment:"Filename"`
+	Extension string `comment:"File extension"`
+	Mime      string `comment:"File mime type"`
+}
+
+func exprEnv(ctx context.Context, file *File) Env {
+	if file == nil {
+		return Env{}
+	}
+
+	var extension = filepath.Ext(file.File)
+	var filename = strings.TrimSuffix(filepath.Base(file.File), extension)
+	var mime, err = mimetype.DetectFile(file.File)
+	if err != nil {
+		mime = &mimetype.MIME{}
+		logger.From(ctx).Error("detect file mime", zap.Error(err))
+	}
+	return Env{
+		File:      file.File,
+		Thumb:     file.Thumb,
+		Filename:  filename,
+		Extension: extension,
+		Mime:      mime.String(),
+	}
+}
+
+type File struct {
+	File  string
+	Thumb string
 }
 
 type iter struct {
-	files  []*file
-	to     peers.Peer
-	photo  bool
-	remove bool
+	files   []*File
+	to      *vm.Program
+	chat    string
+	topic   int
+	photo   bool
+	remove  bool
+	manager *peers.Manager
 
 	cur  int
 	err  error
 	file uploader.Elem
 }
 
-func newIter(files []*file, to peers.Peer, photo, remove bool) *iter {
+type dest struct {
+	Peer   string
+	Thread int
+}
+
+func newIter(files []*File, to *vm.Program, chat string, topic int, photo, remove bool, manager *peers.Manager) *iter {
 	return &iter{
-		files:  files,
-		to:     to,
-		photo:  photo,
-		remove: remove,
+		files:   files,
+		to:      to,
+		chat:    chat,
+		topic:   topic,
+		photo:   photo,
+		remove:  remove,
+		manager: manager,
 
 		cur:  0,
 		err:  nil,
@@ -56,21 +102,61 @@ func (i *iter) Next(ctx context.Context) bool {
 	cur := i.files[i.cur]
 	i.cur++
 
-	f, err := os.Open(cur.file)
+	f, err := os.Open(cur.File)
 	if err != nil {
 		i.err = errors.Wrap(err, "open file")
 		return false
 	}
 
+	var (
+		to     peers.Peer
+		thread int
+	)
+	if i.chat != "" {
+		to, i.err = i.resolvePeer(ctx, i.chat)
+		thread = i.topic
+		if i.err != nil {
+			return false
+		}
+	} else {
+		// message routing
+		result, err := texpr.Run(i.to, exprEnv(ctx, cur))
+		if err != nil {
+			i.err = errors.Wrap(err, "message routing")
+			return false
+		}
+
+		switch r := result.(type) {
+		case string:
+			// pure chat, no reply to, which is a compatible with old version
+			// and a convenient way to send message to self
+			to, err = i.resolvePeer(ctx, r)
+		case map[string]interface{}:
+			// chat with reply to topic or message
+			var d dest
+
+			if err = mapstructure.WeakDecode(r, &d); err != nil {
+				i.err = errors.Wrapf(err, "decode dest: %v", result)
+				return false
+			}
+
+			to, err = i.resolvePeer(ctx, d.Peer)
+			thread = d.Thread
+		default:
+			i.err = errors.Errorf("message router must return string or dest: %T", result)
+			return false
+		}
+	}
+
 	var thumb *uploaderFile = nil
 	// has thumbnail
-	if cur.thumb != "" {
-		tMime, err := mimetype.DetectFile(cur.thumb)
+	if cur.Thumb != "" {
+		tMime, err := mimetype.DetectFile(cur.Thumb)
 		if err != nil || !utils.Media.IsImage(tMime.String()) { // TODO(iyear): jpg only
-			i.err = errors.Wrapf(err, "invalid thumbnail file: %v", cur.thumb)
+			i.err = errors.Wrapf(err, "invalid thumbnail file: %v", cur.Thumb)
 			return false
 		}
-		thumbFile, err := os.Open(cur.thumb)
+		thumbFile, err := os.Open(cur.Thumb)
 		if err != nil {
 			i.err = errors.Wrap(err, "open thumbnail file")
 			return false
@@ -86,17 +172,25 @@ func (i *iter) Next(ctx context.Context) bool {
 	}
 
 	i.file = &iterElem{
-		file:  &uploaderFile{File: f, size: stat.Size()},
-		thumb: thumb,
-		to:    i.to,
-
+		file:    &uploaderFile{File: f, size: stat.Size()},
+		thumb:   thumb,
+		to:      to,
+		thread:  thread,
 		asPhoto: i.photo,
 		remove:  i.remove,
 	}
 
 	return true
 }
 
+func (i *iter) resolvePeer(ctx context.Context, peer string) (peers.Peer, error) {
+	if peer == "" { // self
+		return i.manager.Self(ctx)
+	}
+
+	return utils.Telegram.GetInputPeer(ctx, i.manager, peer)
+}
+
 func (i *iter) Value() uploader.Elem {
 	return i.file
 }

app/up/up.go

@@ -2,12 +2,16 @@ package up
 
 import (
 	"context"
+	"fmt"
+	"github.com/expr-lang/expr"
+	"github.com/expr-lang/expr/vm"
+	"github.com/iyear/tdl/pkg/texpr"
+	"os"
 
 	"github.com/fatih/color"
 	"github.com/go-faster/errors"
 	"github.com/gotd/td/telegram"
 	"github.com/gotd/td/telegram/peers"
-	"github.com/gotd/td/tg"
 	"github.com/spf13/viper"
 	"go.uber.org/multierr"
 
@@ -23,13 +27,27 @@ import (
 
 type Options struct {
 	Chat     string
+	Thread   int
+	To       string
 	Paths    []string
 	Excludes []string
 	Remove   bool
 	Photo    bool
+	Caption  string
 }
 
 func Run(ctx context.Context, c *telegram.Client, kvd kv.KV, opts Options) (rerr error) {
+	if opts.To == "-" {
+		fg := texpr.NewFieldsGetter(nil)
+
+		fields, err := fg.Walk(exprEnv(nil, nil))
+		if err != nil {
+			return fmt.Errorf("failed to walk fields: %w", err)
+		}
+
+		fmt.Print(fg.Sprint(fields, true))
+		return nil
+	}
 	files, err := walk(opts.Paths, opts.Excludes)
 	if err != nil {
 		return err
@@ -44,7 +62,7 @@ func Run(ctx context.Context, c *telegram.Client, kvd kv.KV, opts Options) (rerr
 
 	manager := peers.Options{Storage: storage.NewPeers(kvd)}.Build(pool.Default(ctx))
 
-	to, err := resolveDestPeer(ctx, manager, opts.Chat)
+	to, err := resolveDest(ctx, manager, opts.To)
 	if err != nil {
 		return errors.Wrap(err, "get target peer")
 	}
@@ -57,7 +75,7 @@ func Run(ctx context.Context, c *telegram.Client, kvd kv.KV, opts Options) (rerr
 		Client:   pool.Default(ctx),
 		PartSize: viper.GetInt(consts.FlagPartSize),
 		Threads:  viper.GetInt(consts.FlagThreads),
-		Iter:     newIter(files, to, opts.Photo, opts.Remove),
+		Iter:     newIter(files, to, opts.Chat, opts.Thread, opts.Photo, opts.Remove, manager),
 		Progress: newProgress(upProgress),
 	}
 
@@ -69,10 +87,37 @@ func Run(ctx context.Context, c *telegram.Client, kvd kv.KV, opts Options) (rerr
 	return up.Upload(ctx, viper.GetInt(consts.FlagLimit))
 }
 
-func resolveDestPeer(ctx context.Context, manager *peers.Manager, chat string) (peers.Peer, error) {
-	if chat == "" {
-		return manager.FromInputPeer(ctx, &tg.InputPeerSelf{})
+//func resolveDestPeer(ctx context.Context, manager *peers.Manager, chat string) (peers.Peer, error) {
+//	if chat == "" {
+//		return manager.FromInputPeer(ctx, &tg.InputPeerSelf{})
+//	}
+//
+//	return utils.Telegram.GetInputPeer(ctx, manager, chat)
+//}
+
+// resolveDest parses the input string and returns a vm.Program. It can be a CHAT, a text or a file based on expression engine.
+func resolveDest(ctx context.Context, manager *peers.Manager, input string) (*vm.Program, error) {
+	compile := func(i string) (*vm.Program, error) {
+		// we pass empty peer and message to enable type checking
+		return expr.Compile(i, expr.Env(exprEnv(nil, nil)))
+	}
+
+	// default
+	if input == "" {
+		return compile(`""`)
+	}
+
+	// file
+	if exp, err := os.ReadFile(input); err == nil {
+		return compile(string(exp))
+	}
+
+	// chat
+	if _, err := utils.Telegram.GetInputPeer(ctx, manager, input); err == nil {
+		// convert to const string
+		return compile(fmt.Sprintf(`"%s"`, input))
 	}
 
-	return utils.Telegram.GetInputPeer(ctx, manager, chat)
+	// text
+	return compile(input)
 }

app/up/walk.go

@@ -9,8 +9,8 @@ import (
 	"github.com/iyear/tdl/pkg/utils"
 )
 
-func walk(paths, excludes []string) ([]*file, error) {
-	files := make([]*file, 0)
+func walk(paths, excludes []string) ([]*File, error) {
+	files := make([]*File, 0)
 	excludesMap := map[string]struct{}{
 		consts.UploadThumbExt: {}, // ignore thumbnail files
 	}
@@ -31,10 +31,10 @@ func walk(paths, excludes []string) ([]*file, error) {
 				return nil
 			}
 
-			f := file{file: path}
+			f := File{File: path}
 			t := strings.TrimRight(path, filepath.Ext(path)) + consts.UploadThumbExt
 			if utils.FS.PathExists(t) {
-				f.thumb = t
+				f.Thumb = t
 			}
 
 			files = append(files, &f)

cmd/up.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"github.com/go-faster/errors"
 
 	"github.com/gotd/td/telegram"
 	"github.com/spf13/cobra"
@@ -20,6 +21,12 @@ func NewUpload() *cobra.Command {
 		Short:   "Upload anything to Telegram",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			return tRun(cmd.Context(), func(ctx context.Context, c *telegram.Client, kvd kv.KV) error {
+				if opts.Thread != 0 && opts.Chat == "" {
+					return errors.New("error flags: --chat should be set when --topic is set")
+				}
+				if opts.Chat != "" && opts.To != "" {
+					return errors.New("conflicting flags: --chat and --to cannot be set at the same time")
+				}
 				return up.Run(logger.Named(ctx, "up"), c, kvd, opts)
 			})
 		},
@@ -29,7 +36,9 @@ func NewUpload() *cobra.Command {
 		_chat = "chat"
 		path  = "path"
 	)
-	cmd.Flags().StringVarP(&opts.Chat, _chat, "c", "", "chat id or domain, and empty means 'Saved Messages'")
+	cmd.Flags().StringVarP(&opts.Chat, _chat, "c", "", "chat id or domain, and empty means 'Saved Messages'. Can be used together with --topic flag. Conflicts with --to flag.")
+	cmd.Flags().IntVar(&opts.Thread, "topic", 0, "specify topic id. Must be used together with --chat flag. Conflicts with --to flag.")
+	cmd.Flags().StringVar(&opts.To, "to", "", "destination peer, can be a CHAT or router based on expression engine. Conflicts with --chat and --topic flag.")
 	cmd.Flags().StringSliceVarP(&opts.Paths, path, "p", []string{}, "dirs or files")
 	cmd.Flags().StringSliceVarP(&opts.Excludes, "excludes", "e", []string{}, "exclude the specified file extensions")
 	cmd.Flags().BoolVar(&opts.Remove, "rm", false, "remove the uploaded files after uploading")

pkg/uploader/iter.go

@@ -23,5 +23,6 @@ type Elem interface {
 	File() File
 	Thumb() (File, bool)
 	To() tg.InputPeerClass
+	Thread() int
 	AsPhoto() bool
 }

pkg/uploader/uploader.go

@@ -136,6 +136,7 @@ func (u *Uploader) upload(ctx context.Context, elem Elem) error {
 	_, err = message.NewSender(u.opts.Client).
 		WithUploader(up).
 		To(elem.To()).
+		Reply(elem.Thread()).
 		Media(ctx, media)
 	if err != nil {
 		return errors.Wrap(err, "send message")