package networkmetering

import (
	"compress/gzip"
	"context"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"text/template"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/parquet-go/parquet-go"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/models"
	"github.com/tidbcloud/metering/pkg/common/utils"
	"github.com/tidbcloud/metering/pkg/common/utils/bigquery"
	"github.com/tidbcloud/metering/pkg/common/utils/storage"
	awsobj "github.com/tidbcloud/metering/pkg/common/utils/storage/aws"
	azblob "github.com/tidbcloud/metering/pkg/common/utils/storage/azure"
	"github.com/tidbcloud/metering/pkg/common/utils/utfbom"
	"go.uber.org/zap"
)

const (
	azureBlobNameSuffix   = ".csv.gz"
	awsObjectNameSuffix   = ".csv.gz"
	awsParquertNameSuffix = ".snappy.parquet"
	awsMetaNameSuffix     = "Manifest.json"
	awsJSONNameSuffix     = ".json"
	azureMetaNameSuffix   = "manifest.json"
	billingchachepath     = "billingcache"
	defaultClient         = "default"
	gcptables             = "`gcp-digital-bi.gcp_billing_detailed.gcp_billing_export_resource_v1*`"
)

var (
	errNometaFound = fmt.Errorf("no meta file found")
)

// each hole task will new a NetworkBillingLoad to load data
type NetworkBillingLoad struct {
	conf           *config.ServiceConfig
	objclis        map[string]storage.StorageProvider
	bigquerycli    bigquery.BigQueryProvider
	cachepath      string
	cachebilling   map[string][]*models.AzureBilling
	cacheawsloaded map[string]bool
	premiumResMeta map[string]*premiumReservedMeta
	log            *zap.Logger
}

type premiumReservedMeta struct {
	clusterIDs      map[string]struct{}
	resourcePoolIDs map[string]struct{}
}

func NewNetworkBillingLoad(conf *config.ServiceConfig, log *zap.Logger) *NetworkBillingLoad {
	return &NetworkBillingLoad{
		conf:           conf,
		cachebilling:   make(map[string][]*models.AzureBilling),
		cacheawsloaded: make(map[string]bool),
		premiumResMeta: make(map[string]*premiumReservedMeta),
		objclis:        make(map[string]storage.StorageProvider),
		log:            log.With(zap.String("module", "networkbillingload")),
	}
}

func (n *NetworkBillingLoad) Init() (*NetworkBillingLoad, error) {
	abspath, err := utils.EnsureCache(n.conf.TrafficMeteringConfig.CachePath, billingchachepath)
	if err != nil {
		return n, err
	}
	n.cachepath = abspath
	switch n.conf.Vendor {
	case constants.VendorAzure:
		{
			err := n.ensureAzureClient()
			if err != nil {
				return n, err
			}
		}
	case constants.VendorAWS:
		{
			err := n.ensureAWSClient()
			if err != nil {
				return n, err
			}
		}
	case constants.VendorGCP:
		{
			err := n.ensureGCPClient()
			if err != nil {
				return n, err
			}
		}
	}
	return n, nil
}

func (n *NetworkBillingLoad) ensureAzureClient() error {
	cred, err := azidentity.NewDefaultAzureCredential(nil)
	if err != nil {
		return err
	}
	cli, err := azblob.NewAzureProviderBlob(cred, n.conf.TrafficMeteringConfig.Provider.Azure.BillingProvider.StorageAccountName, n.conf.TrafficMeteringConfig.Provider.Azure.BillingProvider.ContainerName)
	if err != nil {
		return err
	}
	n.objclis[defaultClient] = cli
	return nil
}

func (n *NetworkBillingLoad) ensureAWSClient() error {
	for _, provider := range n.conf.TrafficMeteringConfig.Provider.AWS.BillingProvider {
		cfg, err := awsconfig.LoadDefaultConfig(context.TODO())
		if err != nil {
			return fmt.Errorf("unable to get aws config: %v", err)
		}
		cfg.Region = provider.Region
		var cli *awsobj.AwsProviderS3
		if provider.RoleArn != "" {
			cli = awsobj.NewAwsProviderS3(cfg, provider.BucketName).AssumeRole(context.TODO(), provider.RoleArn)
		} else {
			cli = awsobj.NewAwsProviderS3(cfg, provider.BucketName)
		}
		n.objclis[getAWScliKey(provider.Account, provider.BucketName)] = cli
	}
	storeProvider := n.conf.TrafficMeteringConfig.Provider.AWS.StoreProvider
	storeKey := getAWScliKey(storeProvider.Account, storeProvider.BucketName)
	if _, ok := n.objclis[storeKey]; !ok {
		cfg, err := awsconfig.LoadDefaultConfig(context.TODO())
		if err != nil {
			return fmt.Errorf("unable to get aws config: %v", err)
		}
		cfg.Region = storeProvider.Region
		var storeCli *awsobj.AwsProviderS3
		if storeProvider.RoleArn != "" {
			storeCli = awsobj.NewAwsProviderS3(cfg, storeProvider.BucketName).AssumeRole(context.TODO(), storeProvider.RoleArn)
		} else {
			storeCli = awsobj.NewAwsProviderS3(cfg, storeProvider.BucketName)
		}
		n.objclis[storeKey] = storeCli
	}
	return nil
}

func (n *NetworkBillingLoad) ensureGCPClient() error {
	//gcp will load billing data from bigquery,not need to init storage client
	//init bigquery client
	cli, err := bigquery.NewBigQuery(n.conf.TrafficMeteringConfig.Provider.GCP.BillingProvider.ProjectId, n.conf.TrafficMeteringConfig.Provider.GCP.BillingProvider.Location, n.log)
	if err != nil {
		return err
	}
	n.bigquerycli = cli
	return nil
}

func (n *NetworkBillingLoad) LoadAzure(ctx context.Context, timestamp int64) ([]*models.AzureBilling, error) {
	destDirprefix, dateRange := n.GetCachePathInTime(timestamp)
	if res, ok := n.cachebilling[dateRange]; ok {
		return filterOneDayAzureBilling(res, timestamp), nil
	}
	objcli := n.objclis[defaultClient]
	for _, perfix := range n.conf.TrafficMeteringConfig.Provider.Azure.BillingProvider.Perfix {
		destDir := filepath.Join(destDirprefix, perfix)
		blobs, err := objcli.ListObjectInTime(ctx, timestamp, perfix)
		if err != nil {
			return nil, err
		}
		meta := filterObjectNameSuffix(blobs, azureMetaNameSuffix)
		if len(meta) == 0 {
			return nil, errNometaFound
		}
		err = utils.EnsurePath(destDir)
		if err != nil {
			return nil, err
		}
		// download meta to cache
		err = objcli.DownloadObjects(ctx, meta, destDir)
		if err != nil {
			return nil, err
		}
		// get all parquet files path from meta
		blobfiles, err := ReadAzureMeta(ctx, destDir, meta[0])
		if err != nil {
			return nil, err
		}
		//download all blobs to cache
		err = objcli.DownloadObjects(ctx, blobfiles, destDir)
		if err != nil {
			return nil, err
		}
		//load file to map cache
		err = filepath.Walk(destDir, func(path string, info os.FileInfo, err error) error {
			if err != nil {
				return err
			}
			if !info.IsDir() {
				// skip the file which is not end with .csv.gz
				if !strings.HasSuffix(path, azureBlobNameSuffix) {
					return nil
				}
				one, err := loadCSVFromGzip[models.AzureBilling](n.log, path, &EmptyCSVFilter{})
				if err != nil {
					return err
				}
				n.cachebilling[dateRange] = append(n.cachebilling[dateRange], one...)
			}
			return nil
		})
		if err != nil {
			return nil, err
		}
	}
	return filterOneDayAzureBilling(n.cachebilling[dateRange], timestamp), nil
}

func (n *NetworkBillingLoad) LoadAWS(ctx context.Context, timestamp int64) ([]*models.AWSBilling, error) {
	_, dateRange := n.GetCachePathInTime(timestamp)
	// if res, ok := n.cacheawsbilling[dateRange]; ok {
	// 	return filterOneDayAWSBilling(res, timestamp), nil
	// }
	var result []*models.AWSBilling
	for _, provider := range n.conf.TrafficMeteringConfig.Provider.AWS.BillingProvider {
		var res []*models.AWSBilling
		switch provider.SourceType {
		case constants.AWSBillingSourceTypeCSV:
			{
				var err error
				res, err = n.loadAWSCSV(ctx, timestamp, provider)
				if err != nil {
					return nil, err
				}
			}
		case constants.AWSBillingSourceTypeParquet:
			{
				var err error
				res, err = n.loadAWSParquet(ctx, timestamp, provider)
				if err != nil {
					return nil, err
				}
			}
		}
		if strings.EqualFold(provider.ServicePlan, constants.ServicePlanPremiumReserved) {
			meta, err := n.loadPremiumReservedMeta(ctx)
			if err != nil {
				return nil, err
			}
			for _, one := range res {
				one.MeteringServicePlan = provider.ServicePlan
				one.MeteringProviderAcct = provider.Account
				if isAWSCrossAZOperation(one.LineItemOperation) {
					if matchPremiumReservedBilling(one, meta) {
						one.MeteringPremiumResMatchedOnly = true
					}
				}
			}
		} else {
			for _, one := range res {
				one.MeteringServicePlan = provider.ServicePlan
				one.MeteringProviderAcct = provider.Account
			}
		}
		result = append(result, res...)
	}
	n.cacheawsloaded[dateRange] = true
	return result, nil
}

func (n *NetworkBillingLoad) loadPremiumReservedMeta(ctx context.Context) (*premiumReservedMeta, error) {
	metaPath := n.conf.TrafficMeteringConfig.Provider.AWS.StoreProvider.PremiumReservedMetaS3Path
	storeProvider := n.conf.TrafficMeteringConfig.Provider.AWS.StoreProvider
	cacheKey := "premium_reserved-" + getAWScliKey(storeProvider.Account, storeProvider.BucketName) + "-" + metaPath
	if meta, ok := n.premiumResMeta[cacheKey]; ok {
		return meta, nil
	}
	if strings.TrimSpace(metaPath) == "" {
		return nil, fmt.Errorf("premium_reserved requires storeProvider.premiumReservedMetaS3Path")
	}
	objcliKey := getAWScliKey(storeProvider.Account, storeProvider.BucketName)
	objcli, ok := n.objclis[objcliKey]
	if !ok {
		return nil, fmt.Errorf("premium_reserved metadata client %s not found", objcliKey)
	}
	metaPrefix, err := normalizePremiumReservedMetaPrefix(storeProvider.BucketName, metaPath)
	if err != nil {
		return nil, err
	}
	objects, err := objcli.ListObjectByName(ctx, metaPrefix)
	if err != nil {
		return nil, err
	}
	metaObjects := filterObjectNameSuffix(objects, awsJSONNameSuffix)
	if len(metaObjects) == 0 {
		return nil, fmt.Errorf("no premium_reserved metadata objects found under prefix %s", metaPrefix)
	}
	destDir := filepath.Join(n.cachepath, "premium_reserved_meta", storeProvider.Account)
	if err = utils.EnsurePath(destDir); err != nil {
		return nil, err
	}
	parsed := &premiumReservedMeta{
		clusterIDs:      make(map[string]struct{}),
		resourcePoolIDs: make(map[string]struct{}),
	}
	for i, metaObject := range metaObjects {
		oneDir := filepath.Join(destDir, strconv.Itoa(i))
		if err = utils.EnsurePath(oneDir); err != nil {
			return nil, err
		}
		if err = objcli.DownloadObjects(ctx, []string{metaObject}, oneDir); err != nil {
			return nil, err
		}
		metaPath := filepath.Join(oneDir, filepath.Base(metaObject))
		data, readErr := os.ReadFile(metaPath)
		if readErr != nil {
			return nil, readErr
		}
		oneMeta, parseErr := parsePremiumReservedMeta(data)
		if parseErr != nil {
			return nil, fmt.Errorf("parse premium_reserved metadata %s failed: %w", metaObject, parseErr)
		}
		mergePremiumReservedMeta(parsed, oneMeta)
	}
	if len(parsed.clusterIDs) == 0 && len(parsed.resourcePoolIDs) == 0 {
		return nil, fmt.Errorf("no cluster/resource pool ids found in premium_reserved metadata objects under %s", metaPrefix)
	}
	n.premiumResMeta[cacheKey] = parsed
	n.log.Info("loaded premium_reserved metadata",
		zap.String("account", storeProvider.Account),
		zap.String("bucket", storeProvider.BucketName),
		zap.String("meta_prefix", metaPrefix),
		zap.Int("meta_object_count", len(metaObjects)),
		zap.Int("cluster_count", len(parsed.clusterIDs)),
		zap.Int("resource_pool_count", len(parsed.resourcePoolIDs)),
	)
	return parsed, nil
}

func normalizePremiumReservedMetaPrefix(billingBucket string, s3Path string) (string, error) {
	trimmed := strings.TrimSpace(s3Path)
	if strings.HasPrefix(trimmed, "s3://") {
		withoutSchema := strings.TrimPrefix(trimmed, "s3://")
		parts := strings.SplitN(withoutSchema, "/", 2)
		if len(parts) == 0 || parts[0] == "" {
			return "", fmt.Errorf("invalid premiumReservedMetaS3Path: %s", s3Path)
		}
		if parts[0] != billingBucket {
			return "", fmt.Errorf("premiumReservedMetaS3Path bucket %s does not match configured store bucket %s", parts[0], billingBucket)
		}
		if len(parts) == 1 {
			trimmed = ""
		} else {
			trimmed = parts[1]
		}
	}
	return trimmed, nil
}

func mergePremiumReservedMeta(dst *premiumReservedMeta, src *premiumReservedMeta) {
	if dst == nil || src == nil {
		return
	}
	for id := range src.clusterIDs {
		dst.clusterIDs[id] = struct{}{}
	}
	for id := range src.resourcePoolIDs {
		dst.resourcePoolIDs[id] = struct{}{}
	}
}

func parsePremiumReservedMeta(raw []byte) (*premiumReservedMeta, error) {
	var payload interface{}
	if err := json.Unmarshal(raw, &payload); err != nil {
		return nil, err
	}
	meta := &premiumReservedMeta{
		clusterIDs:      make(map[string]struct{}),
		resourcePoolIDs: make(map[string]struct{}),
	}
	collectPremiumReservedIDs(payload, meta)
	if len(meta.clusterIDs) == 0 && len(meta.resourcePoolIDs) == 0 {
		return nil, fmt.Errorf("no cluster/resource pool ids found in premium_reserved metadata")
	}
	return meta, nil
}

func collectPremiumReservedIDs(payload interface{}, meta *premiumReservedMeta) {
	switch v := payload.(type) {
	case map[string]interface{}:
		for k, val := range v {
			normalizedKey := strings.ToLower(strings.ReplaceAll(k, "-", "_"))
			normalizedKey = strings.ReplaceAll(normalizedKey, " ", "")
			if id, ok := normalizeMetaIDValue(val); ok {
				if isPremiumReservedClusterKey(normalizedKey) {
					meta.clusterIDs[id] = struct{}{}
				}
				if isPremiumReservedResourcePoolKey(normalizedKey) {
					meta.resourcePoolIDs[id] = struct{}{}
				}
			}
			collectPremiumReservedIDs(val, meta)
		}
	case []interface{}:
		for _, one := range v {
			collectPremiumReservedIDs(one, meta)
		}
	}
}

func normalizeMetaIDValue(v interface{}) (string, bool) {
	switch vv := v.(type) {
	case string:
		res := strings.TrimSpace(vv)
		if res == "" {
			return "", false
		}
		return res, true
	case float64:
		return strconv.FormatInt(int64(vv), 10), true
	case int64:
		return strconv.FormatInt(vv, 10), true
	case int:
		return strconv.Itoa(vv), true
	default:
		return "", false
	}
}

func isPremiumReservedClusterKey(key string) bool {
	return key == "tidb_cluster_id" || key == "tidbclusterid" || key == "cluster_id" || key == "clusterid"
}

func isPremiumReservedResourcePoolKey(key string) bool {
	return key == "resource_pool_id" || key == "resourcepoolid" || key == "shared_pool_id" || key == "sharedpoolid"
}

func matchPremiumReservedBilling(billing *models.AWSBilling, meta *premiumReservedMeta) bool {
	if billing == nil || meta == nil {
		return false
	}
	if cluster := strings.TrimSpace(billing.ResourceTagsCluster); cluster != "" {
		if _, ok := meta.clusterIDs[cluster]; ok {
			return true
		}
	}
	if sharedPool := strings.TrimSpace(billing.ResourceTagsSharedPool); sharedPool != "" {
		if _, ok := meta.resourcePoolIDs[sharedPool]; ok {
			return true
		}
	}
	return false
}

func isAWSCrossAZOperation(operation string) bool {
	return strings.HasPrefix(operation, "InterZone-") || strings.HasPrefix(operation, "VPCPeering-")
}

func (n *NetworkBillingLoad) loadAWSCSV(ctx context.Context, timestamp int64, provider config.BillingProviderAWSConfig) ([]*models.AWSBilling, error) {
	destDir, _ := n.GetCachePathInTime(timestamp, provider.Account)
	_, dateRange := n.GetCachePathInTime(timestamp)
	//check if the data has been loaded to local cache
	if _, ok := n.cacheawsloaded[dateRange]; !ok {
		objcli := n.objclis[getAWScliKey(provider.Account, provider.BucketName)]
		objects, err := objcli.ListObjectInTime(ctx, timestamp, provider.Perfix)
		if err != nil {
			return nil, err
		}
		objects = filterObjectNameSuffix(objects, awsObjectNameSuffix)
		//download all blobs to cache

		err = utils.EnsurePath(destDir)
		if err != nil {
			return nil, err
		}
		err = objcli.DownloadObjects(ctx, objects, destDir)
		if err != nil {
			return nil, err
		}
	}
	//load file to map cache
	var result []*models.AWSBilling
	err := filepath.Walk(destDir, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !info.IsDir() {
			// skip the file which is not end with .csv.gz
			if !strings.HasSuffix(path, awsObjectNameSuffix) {
				return nil
			}
			one, err := loadCSVFromGzipStream[models.AWSBilling](n.log, path, &AWSCSVFilter{})
			if err != nil {
				return err
			}
			result = append(result, filterOneDayAWSBilling(one, timestamp)...)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return result, nil
}

func (n *NetworkBillingLoad) loadAWSParquet(ctx context.Context, timestamp int64, provider config.BillingProviderAWSConfig) ([]*models.AWSBilling, error) {
	destDir, _ := n.GetCachePathInTime(timestamp, provider.Account)
	_, dateRange := n.GetCachePathInTime(timestamp)
	// check if the data has been loaded to local cache
	if _, ok := n.cacheawsloaded[dateRange]; !ok {
		objcli := n.objclis[getAWScliKey(provider.Account, provider.BucketName)]
		metas, err := objcli.ListObjectInTime(ctx, timestamp, provider.Perfix)
		if err != nil {
			return nil, err
		}
		meta := filterObjectNameSuffix(metas, awsMetaNameSuffix)
		if len(meta) == 0 {
			return nil, errNometaFound
		}
		err = utils.EnsurePath(destDir)
		if err != nil {
			return nil, err
		}
		// download meta to cache
		err = objcli.DownloadObjects(ctx, meta, destDir)
		if err != nil {
			return nil, err
		}
		// get all parquet files path from meta
		parquetfailes, err := ReadAWSParquetMeta(ctx, destDir, meta[0])
		if err != nil {
			return nil, err
		}
		err = objcli.DownloadObjects(ctx, parquetfailes, destDir)
		if err != nil {
			return nil, err
		}
	}
	//load file to map cache
	var result []*models.AWSBilling
	err := filepath.Walk(destDir, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !info.IsDir() {
			// skip the file which is not end with .parquet
			if !strings.HasSuffix(path, awsParquertNameSuffix) {
				return nil
			}
			one, err := loadParquetStream(n.log, path)
			if err != nil {
				return err
			}
			result = append(result, filterOneDayAWSBilling(one, timestamp)...)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return result, nil
}

func (n *NetworkBillingLoad) GetCachePathInTime(timestamp int64, perfix ...string) (string, string) {
	switch n.conf.Vendor {
	case constants.VendorAzure:
		{
			name := utils.GenerateDateRangeAzure(timestamp)
			return filepath.Join(n.cachepath, name), name
		}
	case constants.VendorAWS:
		{
			name := utils.GenerateDateRangeAWS(timestamp)
			if len(perfix) > 0 {
				return filepath.Join(n.cachepath, perfix[0], name), name
			}
			return filepath.Join(n.cachepath, name), name
		}
	}
	return "", ""
}

func (n *NetworkBillingLoad) LoadGCP(ctx context.Context, timestamp int64) ([]*models.GCPBilling, error) {
	//gcp will load billing data from bigquery
	dateRange := utils.GenerateBillingDateForGCP(timestamp)
	sql, err := GetGCPBillingSQL(gcpsqlParams{
		Tables:             gcptables,
		Projects:           n.conf.TrafficMeteringConfig.Provider.GCP.BillingProvider.MeteringProjects,
		Date:               dateRange,
		PartitionPredicate: buildGCPPartitionPredicate(dateRange, n.conf.TrafficMeteringConfig.RerunDays),
	})
	if err != nil {
		return nil, err
	}
	res, err := n.bigquerycli.ExecQuery(ctx, sql)
	if err != nil {
		return nil, err
	}
	results := make([]*models.GCPBilling, 0)
	for _, r := range res {
		res := &models.GCPBilling{}
		err := Coverage(r, res)
		if err != nil {
			return nil, err
		}
		results = append(results, res)
	}
	return results, nil
}

func loadCSVFromGzip[T any](log *zap.Logger, filepath string, filter CsvLoaderFilter) ([]*T, error) {
	file, err := os.Open(filepath)
	if err != nil {
		return nil, err
	}
	defer file.Close()
	gzReader, err := gzip.NewReader(file)
	if err != nil {
		return nil, err
	}
	defer gzReader.Close()
	sr := utfbom.SkipOnly(gzReader)
	csvReader := csv.NewReader(sr)
	records, err := csvReader.ReadAll()
	if err != nil {
		return nil, err
	}
	if len(records) < 1 {
		return nil, fmt.Errorf("no data found")
	}

	headers := records[0]
	var result = make([]*T, 0)
	var tagmap = make(map[string]int)
	err = filter.Init(headers)
	if err != nil {
		return nil, err
	}
	for _, row := range records[1:] {
		if !filter.Filter(row) {
			continue
		}
		var record T
		v := reflect.ValueOf(&record).Elem()
		t := v.Type()
		//build jsontag map
		if len(tagmap) == 0 {
			for j := 0; j < t.NumField(); j++ {
				field := t.Field(j)
				jsonTag := field.Tag.Get("json")
				tagmap[strings.ToLower(jsonTag)] = j

			}
		}
		for i, header := range headers {
			if j, ok := tagmap[strings.ToLower(header)]; ok {
				f := v.Field(j)
				if f.IsValid() && f.CanSet() {
					switch f.Kind() {
					case reflect.String:
						f.SetString(row[i])
					case reflect.Float64:
						if row[i] == "" {
							f.SetFloat(0)
							break
						}
						val, err := strconv.ParseFloat(row[i], 64)
						if err != nil {
							return nil, err
						}
						f.SetFloat(val)
					case reflect.Bool:
						val, err := strconv.ParseBool(row[i])
						if err != nil {
							return nil, err
						}
						f.SetBool(val)
					}
				}
			}
		}
		result = append(result, &record)
	}
	log.Info("load csv file success", zap.String("path", filepath))
	return result, nil
}

func loadCSVFromGzipStream[T any](log *zap.Logger, filepath string, filter CsvLoaderFilter) ([]*T, error) {
	file, err := os.Open(filepath)
	if err != nil {
		return nil, err
	}
	defer file.Close()
	gzReader, err := gzip.NewReader(file)
	if err != nil {
		return nil, err
	}
	defer gzReader.Close()
	sr := utfbom.SkipOnly(gzReader)
	csvReader := csv.NewReader(sr)
	// records, err := csvReader.ReadAll()
	headers, err := csvReader.Read()
	if err != nil {
		if err == io.EOF {
			return nil, fmt.Errorf("no data found")
		}
		return nil, err
	}
	var result []*T
	var tagmap = make(map[string]int)
	err = filter.Init(headers)
	if err != nil {
		return nil, err
	}
	for row, err := csvReader.Read(); err != io.EOF; row, err = csvReader.Read() {
		if !filter.Filter(row) {
			continue
		}
		var record T
		v := reflect.ValueOf(&record).Elem()
		t := v.Type()
		//build jsontag map
		if len(tagmap) == 0 {
			for j := 0; j < t.NumField(); j++ {
				field := t.Field(j)
				jsonTag := field.Tag.Get("json")
				// jsonTag =
				tagmap[strings.ToLower(jsonTag)] = j

			}
		}
		for i, header := range headers {
			if j, ok := tagmap[strings.ToLower(header)]; ok {
				f := v.Field(j)
				if f.IsValid() && f.CanSet() {
					switch f.Kind() {
					case reflect.String:
						f.SetString(row[i])
					case reflect.Float64:
						if row[i] == "" {
							f.SetFloat(0)
							break
						}
						val, err := strconv.ParseFloat(row[i], 64)
						if err != nil {
							return nil, err
						}
						f.SetFloat(val)
					case reflect.Bool:
						val, err := strconv.ParseBool(row[i])
						if err != nil {
							return nil, err
						}
						f.SetBool(val)
					}
				}
			}
		}
		result = append(result, &record)
	}
	log.Info("load csv file success", zap.String("path", filepath))
	return result, nil
}

func loadParquetStream(log *zap.Logger, filepath string) ([]*models.AWSBilling, error) {
	f, err := os.Open(filepath)
	if err != nil {
		log.Error("Failed to open file", zap.Error(err))
		return nil, err
	}
	defer f.Close()
	schema := parquet.SchemaOf(new(models.AWSBillingParquet))
	r := parquet.NewReader(f, schema)
	var result []*models.AWSBilling
	log.Info("nums", zap.Int64("nums", r.NumRows()))
	for {
		record := new(models.AWSBillingParquet)
		if err := r.Read(record); err != nil {
			if err == io.EOF {
				break
			}
			log.Error("Read error", zap.Error(err))
			return nil, err
		}
		if tmp := filterAWSBillingParquet2AWSBilling(record); tmp != nil {
			result = append(result, tmp)
		}
	}
	log.Info("load parquet file success", zap.String("path", filepath))
	return result, nil
}

func filterAWSBillingParquet2AWSBilling(billing *models.AWSBillingParquet) *models.AWSBilling {
	if billing.LineItemProductCode == "AWSELB" || (billing.LineItemProductCode == "AmazonEC2" && billing.ProductServicename == "AWS Data Transfer") {
		if billing.ResourceTagsComponent == "extra-tidb" || billing.ResourceTagsComponent == "tiflow" || billing.ResourceTagsComponent == "rg_ticdc" {
			return nil
		}
		return &models.AWSBilling{
			IdentityLineItemId:                  billing.IdentityLineItemId,
			IdentityTimeInterval:                billing.IdentityTimeInterval,
			BillInvoiceId:                       billing.BillInvoiceId,
			BillInvoicingEntity:                 billing.BillInvoicingEntity,
			BillBillingEntity:                   billing.BillBillingEntity,
			BillBillType:                        billing.BillBillType,
			BillPayerAccountId:                  billing.BillPayerAccountId,
			LineItemUsageAccountId:              billing.LineItemUsageAccountId,
			LineItemLineItemType:                billing.LineItemLineItemType,
			LineItemUsageStartDate:              time.Unix(0, billing.LineItemUsageStartDate*int64(time.Millisecond)).UTC().Format(time.RFC3339),
			LineItemUsageEndDate:                time.Unix(0, billing.LineItemUsageEndDate*int64(time.Millisecond)).UTC().Format(time.RFC3339),
			LineItemProductCode:                 billing.LineItemProductCode,
			LineItemUsageType:                   billing.LineItemUsageType,
			LineItemOperation:                   billing.LineItemOperation,
			LineItemAvailabilityZone:            billing.LineItemAvailabilityZone,
			LineItemResourceId:                  billing.LineItemResourceId,
			LineItemUsageAmount:                 billing.LineItemUsageAmount,
			LineItemCurrencyCode:                billing.LineItemCurrencyCode,
			LineItemUnblendedCost:               billing.LineItemUnblendedCost,
			ProductRegion:                       billing.ProductRegion,
			ProductRegionCode:                   billing.ProductRegionCode,
			ProductServicename:                  billing.ProductServicename,
			PricingUnit:                         billing.PricingUnit,
			ResourceTagsApplication:             billing.ResourceTagsApplication,
			ResourceTagsDepartment:              billing.ResourceTagsDepartment,
			ResourceTagsEnvironment:             billing.ResourceTagsEnvironment,
			ResourceTagsName:                    billing.ResourceTagsName,
			ResourceTagsOwner:                   billing.ResourceTagsOwner,
			ResourceTagsProject:                 billing.ResourceTagsProject,
			ResourceTagsCluster:                 billing.ResourceTagsCluster,
			ResourceTagsComponent:               billing.ResourceTagsComponent,
			ResourceTagsSharedPool:              billing.ResourceTagsSharedPool,
			ResourceTagsEnv:                     billing.ResourceTagsEnv,
			ResourceTagsKubernetesIoServiceName: billing.ResourceTagsKubernetesIoServiceName,
			ResourceTagsServicetype:             billing.ResourceTagsServicetype,
			ResourceTagsTenant:                  billing.ResourceTagsTenant,
			ResourceTagsUsedby:                  billing.ResourceTagsUsedby,
		}
	}
	return nil
}

func filterObjectNameSuffix(blobs []string, include string) []string {
	var res []string
	for _, blob := range blobs {
		if strings.HasSuffix(blob, include) {
			res = append(res, blob)
		}
	}
	return res
}

func filterOneDayAzureBilling(billings []*models.AzureBilling, timestamp int64) []*models.AzureBilling {
	var res []*models.AzureBilling
	if len(billings) == 0 {
		return res
	}
	day := utils.GenerateBillingDate(timestamp)
	for _, billing := range billings {
		if billing.Date == day {
			res = append(res, billing)
		}
	}
	return res
}

func filterOneDayAWSBilling(billings []*models.AWSBilling, timestamp int64) []*models.AWSBilling {
	var res []*models.AWSBilling
	if len(billings) == 0 {
		return res
	}
	nowday := time.Unix(timestamp, 0).UTC().Day()
	for _, billing := range billings {
		parsedTime, err := time.Parse(time.RFC3339, billing.LineItemUsageStartDate)
		if err != nil {
			fmt.Println("Error parsing time:", err)
			return res
		}
		if parsedTime.Day() == nowday {
			res = append(res, billing)
		}
	}
	return res
}

func getAWScliKey(account string, bucket string) string {
	return fmt.Sprintf("%s-%s", account, bucket)
}

type EmptyCSVFilter struct {
}

func (e *EmptyCSVFilter) Init(headers []string) error {
	return nil
}

func (e *EmptyCSVFilter) Filter(row []string) bool {
	return true
}

type AWSCSVFilter struct {
	productCodeCursor        int
	productServicenameCursor int
	tagComponentCursor       int
}

func (a *AWSCSVFilter) Init(headers []string) error {
	for i, header := range headers {
		if header == "lineItem/ProductCode" {
			a.productCodeCursor = i
		}
		if header == "product/servicename" {
			a.productServicenameCursor = i
		}
		if header == "resourceTags/user:component" {
			a.tagComponentCursor = i
		}
	}
	if a.productCodeCursor == 0 || a.productServicenameCursor == 0 {
		return fmt.Errorf("lineItem/ProductCode %d or product/servicename %d not found", a.productCodeCursor, a.productServicenameCursor)
	}
	return nil
}

func (a *AWSCSVFilter) Filter(row []string) bool {
	if row[a.productCodeCursor] == "AWSELB" || (row[a.productCodeCursor] == "AmazonEC2" && row[a.productServicenameCursor] == "AWS Data Transfer") {
		if row[a.tagComponentCursor] == "extra-tidb" || row[a.tagComponentCursor] == "tiflow" || row[a.tagComponentCursor] == "rg_ticdc" {
			return false
		}
		return true
	}
	return false
}

type gcpsqlParams struct {
	Tables             string
	Projects           []string
	Date               string
	PartitionPredicate string
}

const gcpsqlTmpl = `WITH
billsource AS (
    SELECT
    IFNULL((select value from UNNEST(labels) where key='tenant' limit 1), '') as user_label_tenant_id,
    IFNULL((select value from UNNEST(labels) where key='project' limit 1), '') as user_label_project_id,
    IFNULL((select value from UNNEST(labels) where key='cluster' limit 1), '') as user_label_cluster_id,
    IFNULL((select value from UNNEST(labels) where key='environment' limit 1), '') as user_label_environment_id,
    IFNULL((select value from UNNEST(labels) where key='component' limit 1), '') as user_label_component_id,
    IFNULL((select value from UNNEST(labels) where key='usedby' limit 1), '') as user_label_usedby,
    IFNULL((select value from UNNEST(labels) where key='servicetype' limit 1), '') as user_label_servicetype,
	IFNULL((select value from UNNEST(tags) where key='usedby' LIMIT 1), '') AS user_tag_usedby,
    usage_start_time,usage_end_time,_PARTITIONTIME,
    location.region AS region,sku.description AS description,service.description AS service_name,
    usage.pricing_unit AS unit,usage.amount_in_pricing_units AS amount,cost,cost_at_list,resource.Name as resource_name,CAST(usage_start_time AS DATE) AS usage_start_date
    FROM
      {{ .Tables }}
    WHERE
        project.id IN ({{ range $index, $element := .Projects }}{{ if $index }}, {{ end }}"{{$element}}"{{ end }})
    {{- if .PartitionPredicate }}
    -- _PARTITIONTIME is an arrival-time proxy; keep usage_start_date as the business-day filter.
    AND {{ .PartitionPredicate }}
    {{- end }}
    AND cost > 0
    AND (
        (
            (sku.description LIKE "Network Internet %" OR sku.description LIKE "Network Inter Region %" 
            OR sku.description LIKE "Network Inter Zone %") 
            AND service.description = "Compute Engine" OR (sku.description LIKE "%Network Load Balancer%" AND service.description = "Networking")
        )
    )
)
SELECT
    region,sum(amount) as amount,unit,description,service_name,sum(cost) as cost,sum(cost_at_list) as cost_at_list,
    user_label_tenant_id,user_label_project_id,user_label_cluster_id,user_label_environment_id,user_label_component_id,
    user_label_usedby,user_tag_usedby,user_label_servicetype,usage_start_date,resource_name
FROM
    billsource
WHERE
    (user_label_usedby LIKE "customer%" OR user_tag_usedby = "customer-tidb")
    AND user_label_component_id != "extra-tidb"
    AND user_label_component_id != "tiflow"
    AND user_label_component_id != "rg_ticdc"
    AND usage_start_date = DATE('{{ .Date }}') 
GROUP BY
    region,unit,description,service_name,
    user_label_tenant_id,user_label_project_id,user_label_cluster_id,user_label_environment_id,user_label_component_id,
    user_label_usedby,user_tag_usedby,user_label_servicetype,usage_start_date,resource_name`

func buildGCPPartitionPredicate(date string, rerunDays int) string {
	return fmt.Sprintf(
		"_PARTITIONTIME >= TIMESTAMP(DATE '%s') AND _PARTITIONTIME < TIMESTAMP(DATE_ADD(DATE '%s', INTERVAL %d DAY))",
		date,
		date,
		rerunDays+2,
	)
}

func GetGCPBillingSQL(params gcpsqlParams) (string, error) {
	templ, err := template.New("gcp").Parse(gcpsqlTmpl)
	if err != nil {
		return "", err
	}
	var buf strings.Builder
	err = templ.Execute(&buf, params)
	if err != nil {
		return "", err
	}
	return buf.String(), nil

}

// Coverage An easy way to copy or transform map and structures
func Coverage(in interface{}, addr interface{}) error {
	bytes, err := json.Marshal(in)
	if err != nil {
		return err
	}
	err = json.Unmarshal(bytes, addr)
	if err != nil {
		return err
	}
	return nil
}

func ReadAWSParquetMeta(ctx context.Context, destdir string, metaobj string) ([]string, error) {
	//load meta file
	file, err := os.Open(filepath.Join(destdir, filepath.Base(metaobj)))
	if err != nil {
		return nil, err
	}
	defer file.Close()
	var meta models.AWSParquetMeta
	err = json.NewDecoder(file).Decode(&meta)
	if err != nil {
		return nil, err
	}
	return meta.ReportKeys, nil

}

func ReadAzureMeta(ctx context.Context, destdir string, metaobj string) ([]string, error) {
	//load meta file
	file, err := os.Open(filepath.Join(destdir, filepath.Base(metaobj)))
	if err != nil {
		return nil, err
	}
	defer file.Close()
	var meta models.AzureMeta
	err = json.NewDecoder(file).Decode(&meta)
	if err != nil {
		return nil, err
	}
	var result = make([]string, 0)
	for i := range meta.Blobs {
		result = append(result, meta.Blobs[i].BlobName)
	}
	return result, nil

}