<commit_before>package bucket

import (
	"bufio"
	"github.com/qiniu/qshell/v2/iqshell/common/alert"
	"github.com/qiniu/qshell/v2/iqshell/common/data"
	"github.com/qiniu/qshell/v2/iqshell/common/file"
	"github.com/qiniu/qshell/v2/iqshell/common/log"
	"github.com/qiniu/qshell/v2/iqshell/common/workspace"
	"github.com/qiniu/qshell/v2/iqshell/storage/bucket/internal/list"
	"io"
	"math"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

type ListApiInfo struct {
	Bucket             string    // 空间名	【必选】
	Prefix             string    // 前缀
	Marker             string    // 标记
	Delimiter          string    //
	StartTime          time.Time // list item 的 put time 区间的开始时间 【闭区间】
	EndTime            time.Time // list item 的 put time 区间的终止时间 【闭区间】
	Suffixes           []string  // list item 必须包含后缀
	StorageTypes       []int     // list item 存储类型，多个使用逗号隔开， 0:普通存储 1:低频存储 2:归档存储 3:深度归档存储
	MimeTypes          []string  // list item Mimetype类型，多个使用逗号隔开
	MinFileSize        int64     // 文件最小值，单位: B
	MaxFileSize        int64     // 文件最大值，单位: B
	MaxRetry           int       // -1: 无限重试
	ShowFields         []string  // 需要展示的字段  【必选】
	ApiVersion         string    // list api 版本，v1 / v2【可选】
	V1Limit            int       // 每次请求 size ，list v1 特有
	OutputLimit        int       // 最大输出条数，默认：-1, 无限输出
	OutputFieldsSep    string    // 输出信息，每行的分隔符 【必选】
	OutputFileMaxLines int64     // 输出文件的最大行数，超过则自动创建新的文件，0：不限制输出文件的行数 【可选】
	OutputFileMaxSize  int64     // 输出文件的最大 Size，超过则自动创建新的文件，0：不限制输出文件的大小 【可选】
	EnableRecord       bool      // 是否开启 record 记录，开启后会记录 list 信息，下次 list 会自动指定 Marker 继续 list 【可选】
	CacheDir           string    // 历史数据存储路径 【内部使用】
}

func ListObjectField(field string) string {
	for _, f := range listObjectFields {
		if strings.EqualFold(field, f) {
			return f
		}
	}
	return ""
}

type ListObject = list.Item

// List list 某个 bucket 所有的文件
func List(info ListApiInfo,
	objectHandler func(marker string, object ListObject) (shouldContinue bool, err *data.CodeError),
	errorHandler func(marker string, err *data.CodeError)) {
	if objectHandler == nil {
		log.Error(alert.CannotEmpty("list bucket: object handler", ""))
		return
	}

	if errorHandler == nil {
		errorHandler = func(marker string, err *data.CodeError) {
			log.ErrorF("marker: %s", info.Marker)
			log.ErrorF("list bucket Error: %v", err)
		}
		log.Warning("list bucket: not set error handler")
	}

	bucketManager, err := GetBucketManager()
	if err != nil {
		errorHandler("", err)
		return
	}

	log.Debug("will list bucket")
	log.DebugF("Suffixes:%s", info.Suffixes)
	shouldCheckPutTime := !info.StartTime.IsZero() || !info.EndTime.IsZero()
	shouldCheckSuffixes := len(info.Suffixes) > 0
	shouldCheckStorageTypes := len(info.StorageTypes) > 0
	shouldCheckMimeTypes := len(info.MimeTypes) > 0
	shouldCheckFileSize := info.MinFileSize > 0 || info.MaxFileSize > 0
	isItemExcepted := func(listItem list.Item) (isExcepted bool) {
		if shouldCheckPutTime {
			putTime := time.Unix(listItem.PutTime/1e7, 0)
			if !filterByPutTime(putTime, info.StartTime, info.EndTime) {
				log.DebugF("filter %s: putTime not match, %s out of range [start:%s ~ end:%s]", listItem.Key, putTime, info.StartTime, info.EndTime)
				return false
			}
		}

		if shouldCheckSuffixes && !filterBySuffixes(listItem.Key, info.Suffixes) {
			log.DebugF("filter %s: key not match, key:%s suffixes:%s ", listItem.Key, listItem.Key, info.Suffixes)
			return false
		}

		if shouldCheckStorageTypes && !filterByStorageType(listItem.Type, info.StorageTypes) {
			log.DebugF("filter %s: key not match, storageType:%d StorageTypes:%s ", listItem.Key, listItem.Type, info.Suffixes)
			return false
		}

		if shouldCheckMimeTypes && !filterByMimeType(listItem.MimeType, info.MimeTypes) {
			log.DebugF("filter %s: key not match, mimeType:%s mimeTypes:%s ", listItem.Key, listItem.MimeType, info.MimeTypes)
			return false
		}

		if shouldCheckFileSize && !filterByFileSize(listItem.Fsize, info.MinFileSize, info.MaxFileSize) {
			log.DebugF("filter %s: key not match, fileSize:%d minSize:%d maxSize:%d", listItem.Key, listItem.Fsize, info.MinFileSize, info.MaxFileSize)
			return false
		}

		return true
	}

	listWaiter := sync.WaitGroup{}
	listWaiter.Add(1)
	workspace.AddCancelObserver(func(s os.Signal) {
		listWaiter.Wait()
	})

	cache := &listCache{
		enableRecord: info.EnableRecord,
		cachePath:    filepath.Join(info.CacheDir, "info.json"),
	}
	cacheInfoP, err := cache.loadCache()
	if err != nil {
		log.Debug(err)
	}

	if cacheInfoP != nil && len(cacheInfoP.Marker) > 0 {
		if len(info.Marker) == 0 {
			info.Marker = cacheInfoP.Marker
		}
	}

	if cacheInfoP == nil {
		cacheInfoP = &cacheInfo{}
	} else if len(cacheInfoP.Marker) > 0 {
		log.InfoF("use marker:%s", cacheInfoP.Marker)
	}

	retryCount := 0
	outputCount := 0
	complete := false
	for !complete && (info.MaxRetry < 0 || retryCount <= info.MaxRetry) {
		hasMore, lErr := list.ListBucket(workspace.GetContext(), list.ApiInfo{
			Manager:    bucketManager,
			ApiVersion: list.ApiVersion(info.ApiVersion),
			Bucket:     info.Bucket,
			Prefix:     info.Prefix,
			Delimiter:  info.Delimiter,
			Marker:     info.Marker,
			V1Limit:    info.V1Limit,
		}, func(marker string, dir string, listItem list.Item) (stop bool) {
			if marker != info.Marker {
				info.Marker = marker
			}

			if !isItemExcepted(listItem) {
				return false
			}

			shouldContinue, hErr := objectHandler(marker, listItem)
			if hErr != nil {
				errorHandler(marker, hErr)
			}
			if !shouldContinue {
				complete = true
				return true
			}

			outputCount++
			if info.OutputLimit > 0 && outputCount >= info.OutputLimit {
				complete = true
				return true
			}

			return false
		})

		// 保存信息
		cacheInfoP.Bucket = info.Bucket
		cacheInfoP.Prefix = info.Prefix
		cacheInfoP.Marker = info.Marker
		_ = cache.saveCache(cacheInfoP)

		if workspace.IsCmdInterrupt() && lErr == nil {
			lErr = data.NewError(0, "list is interrupted")
		}

		if lErr != nil || workspace.IsCmdInterrupt() {
			errorHandler(info.Marker, lErr)

			if workspace.IsCmdInterrupt() || // 取消
				strings.Contains(lErr.Error(), "no such bucket") || // 空间不存在，直接结束
				strings.Contains(lErr.Error(), "incorrect zone") || // 空间不正确
				strings.Contains(lErr.Error(), "context canceled") { // 取消
				break
			}
<commit_msg>list bucket: api v1 invalid list limit not retry<commit_after>package bucket

import (
	"bufio"
	"github.com/qiniu/qshell/v2/iqshell/common/alert"
	"github.com/qiniu/qshell/v2/iqshell/common/data"
	"github.com/qiniu/qshell/v2/iqshell/common/file"
	"github.com/qiniu/qshell/v2/iqshell/common/log"
	"github.com/qiniu/qshell/v2/iqshell/common/workspace"
	"github.com/qiniu/qshell/v2/iqshell/storage/bucket/internal/list"
	"io"
	"math"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

type ListApiInfo struct {
	Bucket             string    // 空间名	【必选】
	Prefix             string    // 前缀
	Marker             string    // 标记
	Delimiter          string    //
	StartTime          time.Time // list item 的 put time 区间的开始时间 【闭区间】
	EndTime            time.Time // list item 的 put time 区间的终止时间 【闭区间】
	Suffixes           []string  // list item 必须包含后缀
	StorageTypes       []int     // list item 存储类型，多个使用逗号隔开， 0:普通存储 1:低频存储 2:归档存储 3:深度归档存储
	MimeTypes          []string  // list item Mimetype类型，多个使用逗号隔开
	MinFileSize        int64     // 文件最小值，单位: B
	MaxFileSize        int64     // 文件最大值，单位: B
	MaxRetry           int       // -1: 无限重试
	ShowFields         []string  // 需要展示的字段  【必选】
	ApiVersion         string    // list api 版本，v1 / v2【可选】
	V1Limit            int       // 每次请求 size ，list v1 特有
	OutputLimit        int       // 最大输出条数，默认：-1, 无限输出
	OutputFieldsSep    string    // 输出信息，每行的分隔符 【必选】
	OutputFileMaxLines int64     // 输出文件的最大行数，超过则自动创建新的文件，0：不限制输出文件的行数 【可选】
	OutputFileMaxSize  int64     // 输出文件的最大 Size，超过则自动创建新的文件，0：不限制输出文件的大小 【可选】
	EnableRecord       bool      // 是否开启 record 记录，开启后会记录 list 信息，下次 list 会自动指定 Marker 继续 list 【可选】
	CacheDir           string    // 历史数据存储路径 【内部使用】
}

func ListObjectField(field string) string {
	for _, f := range listObjectFields {
		if strings.EqualFold(field, f) {
			return f
		}
	}
	return ""
}

type ListObject = list.Item

// List list 某个 bucket 所有的文件
func List(info ListApiInfo,
	objectHandler func(marker string, object ListObject) (shouldContinue bool, err *data.CodeError),
	errorHandler func(marker string, err *data.CodeError)) {
	if objectHandler == nil {
		log.Error(alert.CannotEmpty("list bucket: object handler", ""))
		return
	}

	if errorHandler == nil {
		errorHandler = func(marker string, err *data.CodeError) {
			log.ErrorF("marker: %s", info.Marker)
			log.ErrorF("list bucket Error: %v", err)
		}
		log.Warning("list bucket: not set error handler")
	}

	bucketManager, err := GetBucketManager()
	if err != nil {
		errorHandler("", err)
		return
	}

	log.Debug("will list bucket")
	log.DebugF("Suffixes:%s", info.Suffixes)
	shouldCheckPutTime := !info.StartTime.IsZero() || !info.EndTime.IsZero()
	shouldCheckSuffixes := len(info.Suffixes) > 0
	shouldCheckStorageTypes := len(info.StorageTypes) > 0
	shouldCheckMimeTypes := len(info.MimeTypes) > 0
	shouldCheckFileSize := info.MinFileSize > 0 || info.MaxFileSize > 0
	isItemExcepted := func(listItem list.Item) (isExcepted bool) {
		if shouldCheckPutTime {
			putTime := time.Unix(listItem.PutTime/1e7, 0)
			if !filterByPutTime(putTime, info.StartTime, info.EndTime) {
				log.DebugF("filter %s: putTime not match, %s out of range [start:%s ~ end:%s]", listItem.Key, putTime, info.StartTime, info.EndTime)
				return false
			}
		}

		if shouldCheckSuffixes && !filterBySuffixes(listItem.Key, info.Suffixes) {
			log.DebugF("filter %s: key not match, key:%s suffixes:%s ", listItem.Key, listItem.Key, info.Suffixes)
			return false
		}

		if shouldCheckStorageTypes && !filterByStorageType(listItem.Type, info.StorageTypes) {
			log.DebugF("filter %s: key not match, storageType:%d StorageTypes:%s ", listItem.Key, listItem.Type, info.Suffixes)
			return false
		}

		if shouldCheckMimeTypes && !filterByMimeType(listItem.MimeType, info.MimeTypes) {
			log.DebugF("filter %s: key not match, mimeType:%s mimeTypes:%s ", listItem.Key, listItem.MimeType, info.MimeTypes)
			return false
		}

		if shouldCheckFileSize && !filterByFileSize(listItem.Fsize, info.MinFileSize, info.MaxFileSize) {
			log.DebugF("filter %s: key not match, fileSize:%d minSize:%d maxSize:%d", listItem.Key, listItem.Fsize, info.MinFileSize, info.MaxFileSize)
			return false
		}

		return true
	}

	listWaiter := sync.WaitGroup{}
	listWaiter.Add(1)
	workspace.AddCancelObserver(func(s os.Signal) {
		listWaiter.Wait()
	})

	cache := &listCache{
		enableRecord: info.EnableRecord,
		cachePath:    filepath.Join(info.CacheDir, "info.json"),
	}
	cacheInfoP, err := cache.loadCache()
	if err != nil {
		log.Debug(err)
	}

	if cacheInfoP != nil && len(cacheInfoP.Marker) > 0 {
		if len(info.Marker) == 0 {
			info.Marker = cacheInfoP.Marker
		}
	}

	if cacheInfoP == nil {
		cacheInfoP = &cacheInfo{}
	} else if len(cacheInfoP.Marker) > 0 {
		log.InfoF("use marker:%s", cacheInfoP.Marker)
	}

	retryCount := 0
	outputCount := 0
	complete := false
	for !complete && (info.MaxRetry < 0 || retryCount <= info.MaxRetry) {
		var hasMore = false
		var lErr *data.CodeError = nil

		if !workspace.IsCmdInterrupt() {
			hasMore, lErr = list.ListBucket(workspace.GetContext(), list.ApiInfo{
				Manager:    bucketManager,
				ApiVersion: list.ApiVersion(info.ApiVersion),
				Bucket:     info.Bucket,
				Prefix:     info.Prefix,
				Delimiter:  info.Delimiter,
				Marker:     info.Marker,
				V1Limit:    info.V1Limit,
			}, func(marker string, dir string, listItem list.Item) (stop bool) {
				if marker != info.Marker {
					info.Marker = marker
				}

				if !isItemExcepted(listItem) {
					return false
				}

				shouldContinue, hErr := objectHandler(marker, listItem)
				if hErr != nil {
					errorHandler(marker, hErr)
				}
				if !shouldContinue {
					complete = true
					return true
				}

				outputCount++
				if info.OutputLimit > 0 && outputCount >= info.OutputLimit {
					complete = true
					return true
				}

				return false
			})
		}

		// 保存信息
		cacheInfoP.Bucket = info.Bucket
		cacheInfoP.Prefix = info.Prefix
		cacheInfoP.Marker = info.Marker
		_ = cache.saveCache(cacheInfoP)

		if workspace.IsCmdInterrupt() && lErr == nil {
			lErr = data.NewError(0, "list is interrupted")
		}

		if lErr != nil || workspace.IsCmdInterrupt() {
			errorHandler(info.Marker, lErr)

			if workspace.IsCmdInterrupt() || // 取消
				strings.Contains(lErr.Error(), "no such bucket") || // 空间不存在，直接结束
				strings.Contains(lErr.Error(), "incorrect zone") || // 空间不正确
				strings.Contains(lErr.Error(), "context canceled") { // 取消
				break
			}

{'fraction_non_alphanumeric': 0.126408700449208, 'fraction_numerical': 0.007171565923240602, 'mean_word_length': 3.208955223880597, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5692683', 'n_tokens_mistral': 5229, 'n_tokens_neox': 4947, 'n_words': 993}