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 "os/signal"
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)
	enc := json.NewEncoder(os.Stdout)
	if opts.Indent {
		enc.SetIndent("", "  ")
	}

	inter := make(chan os.Signal, 1)
	signal.Notify(inter, os.Interrupt)
	go func() {
		<-inter
		os.Stdout.Sync()
		os.Exit(0)
	}()

	finalNum := 0
	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,
			)
		}
		finalNum += len(values)

		if opts.Verbose {
			log.Printf("Got %d %s (%d, page %d)\n",
				len(values), g.GetNameMul(), finalNum, page)
		}

		for _, value := range values {
			err := enc.Encode(value)
			if err != nil {
				log.Fatalf("json.Encode(...): %s\n", err)
			}
		}

		page++
		for page <= opts.EndPage && next != nil {
			values, next, err = next()
			if err != nil {
				log.Fatalf(
					"%s(...): %s\n",
					g.GetFuncName(), err,
				)
			}
			finalNum += len(values)
			if opts.Verbose {
				log.Printf("Got %d %s (%d, page %d)\n",
					len(values), g.GetNameMul(),
					finalNum, page)
			}

			for _, value := range values {
				err = enc.Encode(value)
				if err != nil {
					log.Fatalf("json.Encode(...): %s\n", err)
				}
			}

			page++
		}

	}

	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 {
		finalNum += len(values)
		for _, value := range values {
			err := enc.Encode(value)
			if err != nil {
				log.Fatalf("json.Encode(...): %s\n", err)
			}
		}
		if opts.Verbose {
			log.Printf(
				"thtread(main) %.2f%%, sument=%d",
				float32(finalNum)/float32(len(ids))*100.,
				finalNum,
			)
		}
	}

	if opts.Verbose {
		rm := c.API.RequestsMade()
		log.Printf(
			"Summarized got %d %s\n",
			finalNum, 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)
	}
	
}