amo/cmd/amocli/common.go

package main

import "surdeus.su/core/cli/mtool"
import "surdeus.su/core/amo"
import "sync"
import "os"
import "strconv"
import "log"
import "bufio"
import "math"

type DefaultFlags struct {
	SecretPath string
	Threads int
	Verbose bool
	MRPS int
	MEPR int

	All bool
	StartPage int
	EndPage int
}

func MakeDefaultFlags(opts *DefaultFlags, flags *mtool.Flags) {
	flags.StringVar(
		&opts.SecretPath,
		"secret",
		"secret.json",
		"path to JSON file with AMO CRM secrets",
		"AMO_SECRET",
	)
	flags.IntVar(
		&opts.MRPS,
		"mrps",
		9,
		"maximum requests per second",
	)
	flags.IntVar(
		&opts.Threads,
		"threads",
		5,
		"amount of threads to run the requests",
	)
	flags.BoolVar(
		&opts.Verbose,
		"no-verbose",
		true,
		"disable verbose mode",
	)
	flags.IntVar(
		&opts.MEPR,
		"mepr",
		amo.MEPR,
		"max entities per request",
	)
}

func MakeGetterFlags(
	opts *DefaultFlags,
	flags *mtool.Flags,
) {
	flags.BoolVar(
		&opts.All,
		"all",
		false,
		"get all leads",
	)
	flags.IntVar(
		&opts.StartPage,
		"startpage",
		1,
		"the page to start at (works only with -all)",
	)
	flags.IntVar(
		&opts.EndPage,
		"endpage",
		math.MaxInt,
		"the page to end the requests",
	)
}

// Run function for slice's parts in different threads.
// Returns a channel that ticks that the threads finished.
func RunForSliceInThreads[V any](
	threadN, mepr int, // Thread amount and MERP.
	slice []V,
	// The function that takes
	// the thread number and the slice of slice.
	fn func(int, []V),
) (chan struct{}) {
	ret := make(chan struct{})

	// No need for threads if
	// there are so few entities.
	if len(slice) <= mepr {
		go func() {
			fn(1, slice)
			ret <- struct{}{}
		}()
		return ret
	}

	var wg sync.WaitGroup

	runFn :=func(thread int, s []V) {
		defer wg.Done()
		iterN := (len(s) / mepr) + 1
		for j := 0 ; j<iterN ; j++ {
			start := j*mepr
			end := start + mepr
			if end > len(s) {
				end = len(s)
			}
			if len(s[start:end]) == 0 {
				continue
			}
			fn(thread, s[start:end])
		}
	}
	// Maximizing speed on small data.


	//threadSize := len(slice) / threadN
	threadMeprRest := len(slice) % mepr
	//threadSize := (len(slice)-threadMeprRest)/threadN
	preThreadSize := 
		((len(slice)-threadMeprRest)/threadN)
	threadSize := (preThreadSize/mepr+1)*mepr
	
	if threadSize < mepr {
		threadSize = mepr
		threadN = len(slice) / mepr
		runFn = func(thread int, s []V) {
			defer wg.Done()
			fn(thread, s)
		}
	}

	for i := 0 ; i<threadN ; i++ {
		first := i * threadSize
		last := first + threadSize
		if last > len(slice) {
			last = len(slice)
		}

		// Got an empty slice.
		if len(slice[first:last]) == 0  {
			break
		}

		wg.Add(1)
		go runFn(i+1, slice[first:last])
	}

	go func() {
		wg.Wait()
		ret <- struct{}{}
	}()

	return ret
}

func ReadIDs(idStrs []string, flags *mtool.Flags) []int {
	var ids []int
	if len(idStrs) > 0 {
		ids = make([]int, 0, len(idStrs))
		for _, idStr := range idStrs {
			id, err := strconv.Atoi(idStr)
			if err != nil {
				log.Printf("Error: Atoi(%q): %s\n", err)
				continue
			}
			ids = append(ids, id)
		}
	} else {
		ids = make([]int, 0)
		scanner := bufio.NewScanner(os.Stdin)
		for scanner.Scan() {
			txt := scanner.Text()

			id, err := strconv.Atoi(txt)
			if err != nil {
				log.Printf(
					"strconv.Atoi(%q): %s\n",
					txt, err,
				)
				continue
			}

			ids = append(ids, id)
		}
	}
	return ids
}