amo/cmd/amocli/getter.go

package main

import "surdeus.su/core/amo"
import "surdeus.su/core/ss/urlenc"

import "surdeus.su/core/cli/mtool"

import "fmt"
import "log"
import "time"
import "os"
import "encoding/json"

type Getter[V any] interface {
	GetValues(*amo.Client, ...urlenc.Builder) ([]V, amo.NextFunc[[]V], error)
	GetNameMul() string
	GetFuncName() string
}

func RunGetter[V any, G Getter[V]](g G, flags *mtool.Flags) {
	var (
		opts DefaultFlags
	)

	now := time.Now()
	MakeDefaultFlags(&opts, flags)
	MakeGetterFlags(&opts, flags)

	idStrs := flags.Parse()
	c, err := amo.NewClient(opts.SecretPath)
	if err != nil {
		log.Fatalf("NewAmoClient(...): %s\n", err)
	}
	c.API.SetMRPS(opts.MRPS)
	finalValues := []V{}

	if opts.All {
		page := opts.StartPage
		values, next, err := g.GetValues(
			c,
			urlenc.Value[int]{"page", page},
			urlenc.Value[int]{"limit", opts.MEPR},
		)
		if err != nil {
			log.Fatalf(
				"%s(...): %s\n", g.GetFuncName(), err,
			)
		}
		finalValues = append(finalValues, values...)
		if opts.Verbose {
			log.Printf("Got %d %s (%d, page %d)\n",
				len(values), g.GetNameMul(), len(finalValues), page)
		}
		page++
		for page <= opts.EndPage && next != nil {
			values, next, err = next()
			if err != nil {
				log.Fatalf("amo.GetLeads(...): %s\n", err)
			}
			finalValues = append(finalValues, values...)
			if opts.Verbose {
				log.Printf("Got %d %s (%d, page %d)\n",
					len(values), g.GetNameMul(),
					len(finalValues), page)
			}
			page++
		}

		bts, err := json.MarshalIndent(finalValues, "", "  ")
		if err != nil {
			log.Fatalf("json.MarshalIndent(...) %s\n", err)
		}
		os.Stdout.Write(bts)
		return
	}

	ids := ReadIDs(idStrs)
	if len(ids) == 0 {
		log.Fatalf("Got no IDs to read %s", g.GetNameMul())
		return
	}

	valueChan := make(chan []V)
	finish := RunForSliceInThreads[int](
		opts.Threads, opts.MEPR,
		ids, func(thread int, s []int){
			values, _, err := g.GetValues(
				c,
				urlenc.Array[int]{
					"id",
					s,
				},
				urlenc.Value[string]{
					"with",
					"contacts",
				},
				urlenc.Value[int]{
					"limit",
					opts.MEPR,
				},
			)
			if err != nil {
				log.Printf("thread(%d): %s(...): %s\n",
					thread, g.GetFuncName(), err)
				return
			}
			valueChan <- values
			if opts.Verbose {
				log.Printf(
					"thread(%d): Got %d %s\n",
					thread, len(values),
					g.GetNameMul(),
				)
			}
		},
	)

	//var wg sync.WaitGroup
	go func(){
		// Waiting for appending so we do not lose data.
		<-finish
		for len(valueChan) > 0 {}
		close(valueChan)
	}()

	for values := range valueChan {
		finalValues = append(finalValues, values...)
	}

	if opts.Verbose {
		rm := c.API.RequestsMade()
		log.Printf(
			"Summarized got %d %s\n",
			len(finalValues), g.GetNameMul(),
		)
		log.Printf(
			"Made %d requests in process\n",
			rm,
		)
		took := time.Since(now).Seconds()
		log.Printf(
			"Took %f seconds\n",
			took,
		)
		log.Printf("RPS = %f\n", float64(rm)/took)
	}
	bts, err := json.MarshalIndent(finalValues, "", "  ")
	if err != nil {
		log.Fatalf("json.Marshal(...): %s\n", err)
	}
	fmt.Printf("%s\n", bts)
}