package service

import (
	"context"
	"io"
	"os"
	"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"
)

type GcsSrv interface {
	UploadToGCS(ctx context.Context, buckType string) error
}

type gcsService struct {
	store store.Factory
}

var _ GcsSrv = (*gcsService)(nil)

func newGcs(str *service) *gcsService {
	return &gcsService{store: str.store}
}

func (g *gcsService) UploadToGCS(ctx context.Context, cloudName string) error {
	localPath := viper.GetString("localPath")
	// GCS configurations
	gcsServiceAccountFile := viper.GetString("gcs.sa")
	_, err := os.Stat(gcsServiceAccountFile)
	if err != nil {
		log.L(ctx).Errorf("Failed to get file info: %v", err)

		return err
	}

	client, err := storage.NewClient(ctx, option.WithCredentialsFile(gcsServiceAccountFile))
	if err != nil {
		log.L(ctx).Errorf("Failed to create client: %v", err)

		return err
	}
	defer client.Close()

	// filepath.Walk(gcsPath, 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 := strings.TrimPrefix(path, gcsPath+"/")
	// 	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
	// })

	localFile := localPath + "/" + cloudName + "/" + cloudName + ".csv"
	if cloudName == model.AwsCN {
		localFile = localPath + "/" + cloudName + "/" + cloudName + ".parquet"
	}
	file, err := os.Open(localFile)
	if err != nil {
		log.L(ctx).Errorf("Failed to open local file: %v", err)

		return err
	}
	defer file.Close()
	gcsFileName := cloudName + "/" + cloudName + ".csv"
	if cloudName == model.AwsCN {
		gcsFileName = cloudName + "/" + cloudName + ".parquet"
	}
	writer := client.Bucket(model.Bucket).Object(gcsFileName).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 gcs", localFile)

	return nil
}

func (g *gcsService) Upload(ctx context.Context, filePath, gcsPath string) error {
	fileName := strings.Split(filePath, "/")[len(strings.Split(filePath, "/"))-1]
	// GCS configurations
	gcsServiceAccountFile := viper.GetString("gcs.sa")
	_, err := os.Stat(gcsServiceAccountFile)
	if err != nil {
		log.L(ctx).Errorf("Failed to get file info: %v", err)

		return err
	}

	client, err := storage.NewClient(ctx, option.WithCredentialsFile(gcsServiceAccountFile))
	if err != nil {
		log.L(ctx).Errorf("Failed to create client: %v", err)

		return err
	}
	defer client.Close()

	file, err := os.Open(filePath)
	if err != nil {
		log.L(ctx).Errorf("Failed to open local file: %v", err)

		return err
	}
	defer file.Close()
	gcsFileName := gcsPath + "/" + fileName
	writer := client.Bucket(model.Bucket).Object(gcsFileName).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 gcs", fileName)

	return nil
}