package service

import (
	"context"
	"io"
	"os"
	"path/filepath"
	"strings"

	"cloud.google.com/go/storage"
	"github.com/spf13/viper"
	"google.golang.org/api/option"

	"github.com/pingcap-inc/data-ingestion-tools/internal/itcronjob/store"
	"github.com/pingcap-inc/data-ingestion-tools/model"
	"github.com/pingcap-inc/data-ingestion-tools/pkg/log"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
)

type S3Srv interface {
	FetchAwsCN(ctx context.Context) error
	Upload(ctx context.Context) error
}

type s3Service struct {
	store store.Factory
}

var _ S3Srv = (*s3Service)(nil)

func newS3(str *service) *s3Service {
	return &s3Service{store: str.store}
}

const (
	s3bucket = "billing-cur-for-ecv"
)

func (s *s3Service) FetchAwsCN(ctx context.Context) error {
	// 1. 下载 S3 对象到本地
	err := s.DownloadFormS3(ctx)
	if err != nil {
		return err
	}
	err = s.Upload(ctx)
	if err != nil {
		return err
	}
	log.L(ctx).Info("Fetch awscn success")

	return nil
}

func (s *s3Service) Upload(ctx context.Context) error {
	localPath := viper.GetString("localPath")
	client, err := storage.NewClient(ctx, option.WithCredentialsFile(viper.GetString("gcs.sa")))
	if err != nil {
		log.L(ctx).Errorf("Failed to create client: %v", err)

		return err
	}
	defer client.Close()

	filepath.Walk(localPath+"/awscn", func(path string, info os.FileInfo, err error) error {
		// 如果当前路径是一个目录，则跳过
		if info.IsDir() {
			return nil
		}

		// 打开本地文件
		file, err := os.Open(path)
		if err != nil {
			log.L(ctx).Errorf("Failed to open local file: %v", err)

			return err
		}
		defer file.Close()
		// 设置上传参数
		fileName := "awscn/" + info.Name()
		writer := client.Bucket(model.Bucket).Object(fileName).NewWriter(ctx)
		if _, err = io.Copy(writer, file); err != nil {
			log.L(ctx).Errorf("Failed to upload file: %v", err)

			return err
		}
		if err := writer.Close(); err != nil {
			log.L(ctx).Errorf("Failed to close writer: %v", err)

			return err
		}
		log.L(ctx).Infof("Uploaded %s to %s", path, fileName)

		return nil
	})

	return nil
}

func (s *s3Service) DownloadFormS3(ctx context.Context) error {
	localPath := viper.GetString("localPath")
	key := viper.GetString("s3.key")
	secret := viper.GetString("s3.secret")
	sess2, err := session.NewSession(&aws.Config{
		Credentials: credentials.NewStaticCredentialsFromCreds(credentials.Value{
			AccessKeyID:     key,
			SecretAccessKey: secret,
		}),
		Region: aws.String(model.Region),
	})
	if err != nil {
		log.L(ctx).Errorf("create session error: %v", err)

		return err
	}
	svc := s3.New(sess2)

	downloader := s3manager.NewDownloaderWithClient(svc)

	// 打开本地文件夹
	localDir, err := os.Open(localPath + "/" + model.AwsCN)
	if err != nil {
		log.L(ctx).Errorf("Failed to open local file: %v", err)

		return err
	}
	defer localDir.Close()

	// 遍历 S3 存储桶中的所有对象
	err = svc.ListObjectsPages(&s3.ListObjectsInput{
		Bucket: aws.String(s3bucket),
	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
		for _, obj := range page.Contents {
			// 构建 S3 对象键
			s3key := *obj.Key

			split := strings.Split(s3key, "/")

			fileName := split[len(split)-3][5:] + "-" + split[len(split)-2][6:] + "-"
			if strings.Contains(split[len(split)-1], "001") {
				fileName += "a.parquet"
			} else {
				fileName += "b.parquet"
			}
			// 下载 S3 对象到本地文件
			file, err := os.Create(localPath + "/" + model.AwsCN + "/" + fileName)
			if err != nil {
				log.L(ctx).Errorf("Failed to create local file: %v", err)
			}

			_, err = downloader.Download(file, &s3.GetObjectInput{
				Bucket: aws.String(s3bucket),
				Key:    aws.String(s3key),
			})
			if err != nil {
				log.L(ctx).Errorf("Failed to download file: %v", err)
			}

			log.L(ctx).Infof("Downloaded %s to %s", s3key, fileName)

			// 关闭本地文件
			file.Close()
		}
		return !lastPage
	})
	if err != nil {
		log.L(ctx).Errorf("Failed to list objects: %v", err)

		return err
	}

	return nil
}

// func MergeParquet(ctx context.Context) error {
//
// 	client := http.Client{}
// 	req, err := http.NewRequest("GET", "http://ditool-service.ditools.svc.cluster.local:5000/merge", nil)
// 	if err != nil {
// 		log.L(ctx).Errorf("Failed to create request: %v", err)
//
// 		return err
// 	}
//
// 	res, err := client.Do(req)
// 	if err != nil {
// 		log.L(ctx).Errorf("Failed to send request: %v", err)
//
// 		return err
// 	}
// 	var message interface{}
// 	err = json.NewDecoder(res.Body).Decode(&message)
// 	if err != nil {
//
// 	}
// 	log.L(ctx).Infof("Merge parquet success: %v", message)
//
// 	return nil
// }

// func PushBq(ctx context.Context) {
// 	sa := viper.GetString("gcs.sa")
// 	gcsPrefix := "gs://" + awscnGcs + "/"
// 	client, err := bigquery.NewClient(ctx, gcpProject, option.WithCredentialsFile(sa))
// 	sClient, err := storage.NewClient(ctx, option.WithCredentialsFile(sa))
// 	if err != nil {
// 		log.L(ctx).Errorf("Failed to create client: %v", err)
//
// 		return
// 	}
// 	defer client.Close()
//
// 	it := sClient.Bucket(awscnGcs).Objects(ctx, nil)
// 	for {
// 		object, err := it.Next()
// 		if err == iterator.Done {
// 			break
// 		}
// 		if err != nil {
// 			log.L(ctx).Errorf("Failed to get next object: %v", err)
//
// 			return
// 		}
// 		gcsRf := bigquery.NewGCSReference(gcsPrefix + object.Name)
// 		gcsRf.SourceFormat = bigquery.Parquet
// 		gcsRf.AutoDetect = true
// 		loader := client.Dataset(dataset).Table(strings.ReplaceAll(object.Name, ".parquet", "")).LoaderFrom(gcsRf)
// 		loader.WriteDisposition = bigquery.WriteTruncate
//
// 		job, err := loader.Run(ctx)
// 		if err != nil {
// 			log.L(ctx).Errorf("Failed to run loader: %v", err)
//
// 			return
// 		}
// 		status, err := job.Wait(ctx)
// 		if err != nil {
// 			log.L(ctx).Errorf("Failed to wait for job: %v", err)
//
// 			return
// 		}
// 		if err := status.Err(); err != nil {
// 			log.L(ctx).Errorf("Failed to get job status: %v", err)
//
// 			return
// 		}
// 	}
// }