package service

import (
	"context"
	"encoding/csv"
	"fmt"
	"io"
	"reflect"
	"strconv"
	"strings"

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

	"github.com/pingcap-inc/data-ingestion-tools/model"
	"github.com/pingcap-inc/data-ingestion-tools/pkg/log"
)

func GetBq(ctx context.Context, bqSrv *bigquery.Client, first bool) ([]*model.OrgBq, error) {
	lastMonth := GetLastMonth()
	// querySql12 := "SELECT * FROM `gcp-digital-bi.model_of_monthly_cloud_revrec_confirm.dwm_monthly_order_info_calculation12_of_revrec_tenant_lite_version` where bill_month='" + lastMonth + "'"
	// records12, err := GetBqData(ctx, bqSrv, querySql12)
	// if err != nil {
	// 	return nil, err
	// }
	//
	// if !first {
	// 	return records12, nil
	// }

	querySql11 := "SELECT * FROM `gcp-digital-bi.model_of_monthly_cloud_revrec_confirm.dwm_monthly_order_info_calculation11_of_revrec_tenant_lite_version_v2` where bill_month='" + lastMonth + "'"

	records11, err := GetBqData(ctx, bqSrv, querySql11)
	if err != nil {
		return nil, err
	}

	duplicateMap := make(map[string]bool)
	var addRecords []*model.OrgBq
	// 将重复的数据金额相加,根据adjust_tenant_id和bill_month去重,
	for _, record := range records11 {
		if record.NeedNewOrder.StringVal == "yes" {
			key := record.BillMonth.StringVal + "-" + record.AdjustTenantID.StringVal
			if _, ok := duplicateMap[key]; ok {
				continue
			}
			duplicateMap[key] = true
			addRecords = append(addRecords, record)
		} else {
			addRecords = append(addRecords, record)
		}
	}

	//
	// addRecords := make([]*model.OrgBq, 0)
	// rd11Map := make(map[string]*model.OrgBq)
	// for _, record := range records11 {
	// 	rd11Map[record.BillMonth.StringVal+record.OriginalTenantID.StringVal] = record
	// }
	//
	// for _, record := range records12 {
	// 	key := record.BillMonth.StringVal + record.OriginalTenantID.StringVal
	// 	if rd11, ok := rd11Map[key]; ok {
	// 		if CompareBq(ctx, rd11, record) {
	// 			addRecords = append(addRecords, record)
	// 		} else {
	// 			addRecords = append(addRecords, rd11)
	// 		}
	//
	// 		delete(rd11Map, key)
	// 	}
	// }
	//
	// for _, record := range rd11Map {
	// 	addRecords = append(addRecords, record)
	// }
	//
	// err = WriteBqData(ctx, bqSrv, addRecords)
	// if err != nil {
	// 	return nil, err
	// }

	return addRecords, nil
}

func GetBqData(ctx context.Context, bqSrv *bigquery.Client, querySql string) ([]*model.OrgBq, error) {
	query := bqSrv.Query(querySql)
	it, err := query.Read(ctx)
	if err != nil {
		log.L(ctx).Errorf("Failed to read query: %v", err)

		return nil, err
	}
	res := make([]*model.OrgBq, 0)

	for {
		var value model.OrgBq
		err := it.Next(&value)
		if err == iterator.Done {
			break
		}
		if err != nil {
			log.L(ctx).Errorf("Failed to read row: %v", err)

			return nil, err
		}

		res = append(res, &value)
	}

	return res, nil
}

func CompareBq(ctx context.Context, bqOrder1, bqOrder2 *model.OrgBq) bool {
	if bqOrder1.StartDate.Date != bqOrder2.StartDate.Date ||
		bqOrder1.EndDate.Date != bqOrder2.EndDate.Date ||
		bqOrder1.AccountID.StringVal != bqOrder2.AccountID.StringVal ||
		bqOrder1.AccountName.StringVal != bqOrder2.AccountName.StringVal ||
		bqOrder1.MarketplaceAmount.Float64 != bqOrder2.MarketplaceAmount.Float64 ||
		bqOrder1.StripeAmount.StringVal != bqOrder2.StripeAmount.StringVal ||
		bqOrder1.WireTransferAmount.Float64 != bqOrder2.WireTransferAmount.Float64 ||
		bqOrder1.TotalAmount.Float64 != bqOrder2.TotalAmount.Float64 ||
		bqOrder1.StripeTaxRate.StringVal != bqOrder2.StripeTaxRate.StringVal ||
		bqOrder1.StripeAmountIncludingTax.StringVal != bqOrder2.StripeAmountIncludingTax.StringVal ||
		bqOrder1.StripeInvoiceNumber.StringVal != bqOrder2.StripeInvoiceNumber.StringVal ||
		bqOrder1.StripeInvoice.StringVal != bqOrder2.StripeInvoice.StringVal ||
		bqOrder1.StripeInvoicePaid.StringVal != bqOrder2.StripeInvoicePaid.StringVal ||
		bqOrder1.StripeInvoiceFinalizedDate.StringVal != bqOrder2.StripeInvoiceFinalizedDate.StringVal ||
		bqOrder1.StripeInvoicePaidDate.StringVal != bqOrder2.StripeInvoicePaidDate.StringVal ||
		bqOrder1.InvoiceAmount.Float64 != bqOrder2.InvoiceAmount.Float64 ||
		bqOrder1.NeedNewOrder.StringVal != bqOrder2.NeedNewOrder.StringVal ||
		bqOrder1.NewOrderStartDate.Date != bqOrder2.NewOrderStartDate.Date ||
		bqOrder1.NewOrderEndDate.Date != bqOrder2.NewOrderEndDate.Date ||
		bqOrder1.NewOrderAmount.Float64 != bqOrder2.NewOrderAmount.Float64 ||
		bqOrder1.OrderNumber.StringVal != bqOrder2.OrderNumber.StringVal ||
		bqOrder1.OrderStartDate.Date != bqOrder2.OrderStartDate.Date ||
		bqOrder1.OrderEndDate.Date != bqOrder2.OrderEndDate.Date ||
		bqOrder1.RevenueAmount.Float64 != bqOrder2.RevenueAmount.Float64 ||
		bqOrder1.OrderNumber2.StringVal != bqOrder2.OrderNumber2.StringVal ||
		bqOrder1.OrderStartDate2.Date != bqOrder2.OrderStartDate2.Date ||
		bqOrder1.OrderEndDate2.Date != bqOrder2.OrderEndDate2.Date ||
		bqOrder1.RevenueAmount2.Float64 != bqOrder2.RevenueAmount2.Float64 ||
		bqOrder1.AdjustTenantID.StringVal != bqOrder2.AdjustTenantID.StringVal {
		return false
	}

	return true
}

func WriteBqData(ctx context.Context, bqSrv *bigquery.Client, rows []*model.OrgBq) error {
	log.L(ctx).Infof("write bq data to bigquery")
	if len(rows) == 0 {
		return nil
	}

	lastMonth := GetLastMonth()

	// get csv from gcs
	client, err := storage.NewClient(ctx)
	// client, err := storage.NewClient(ctx, option.WithCredentialsFile(credsFilePath))
	if err != nil {
		return err
	}
	defer client.Close()

	// Create a handle to the bucket
	bucket := client.Bucket(bucketName)
	// Create a handle to the object
	obj := bucket.Object(objectName)

	// Read the object
	r, err := obj.NewReader(ctx)
	if err != nil {
		return err
	}
	defer r.Close()

	// Read CSV contents
	csvReader := csv.NewReader(r)
	records, err := csvReader.ReadAll()
	if err != nil {
		return err
	}

	colIndexMap := make(map[string]int)
	for i, col := range records[0] {
		colIndexMap[col] = i
	}

	rowsMap := make(map[string]*model.OrgBq)
	for _, row := range rows {
		rowsMap[row.BillMonth.StringVal+row.OriginalTenantID.StringVal] = row
	}

	// del bill_month = lastMonth
	delCsv := make([]int, 0)
	for i, record := range records {
		if i == 0 {
			continue
		}
		if record[colIndexMap["bill_month"]] == lastMonth {
			delCsv = append(delCsv, i)
		}
	}

	// from right to left
	for i := len(delCsv) - 1; i >= 0; i-- {
		delIndex := delCsv[i]
		records = append(records[:delIndex], records[delIndex+1:]...)
	}

	addCsv := make([][]string, 0)
	for _, row := range rows {
		fromStruct, err := CsvRecordFromStruct(*row, records[0])
		if err != nil {
			return err
		}
		addCsv = append(addCsv, fromStruct)
	}

	records = append(records, addCsv...)

	var csvContent strings.Builder
	csvWriter := csv.NewWriter(&csvContent)
	if err := csvWriter.WriteAll(records); err != nil {
		log.Fatalf("Failed to write CSV data: %v", err)
	}
	csvWriter.Flush()

	// 创建一个新的io.Reader
	csvData := strings.NewReader(csvContent.String())

	// 创建一个新的GCS writer来上传修改后的文件
	writer := client.Bucket(bucketName).Object(objectName).NewWriter(ctx)
	if _, err := io.Copy(writer, csvData); err != nil {
		log.Fatalf("Failed to write CSV data to GCS: %v", err)
	}

	if err := writer.Close(); err != nil {
		log.Fatalf("Failed to close GCS writer: %v", err)
	}

	schema, _ := SchemaFromFile(viper.GetString("bq.schema"))

	// load csv to bigquery
	gcsRef := bigquery.NewGCSReference(fmt.Sprintf("gs://%s/%s", bucketName, objectName))
	gcsRef.SourceFormat = bigquery.CSV
	gcsRef.AutoDetect = false

	gcsRef.Schema = schema
	gcsRef.FieldDelimiter = ","
	gcsRef.SkipLeadingRows = 1

	loader := bqSrv.Dataset(bqDatasetId).Table(bqTableId).LoaderFrom(gcsRef)
	loader.WriteDisposition = bigquery.WriteTruncate

	job, err := loader.Run(ctx)

	if err != nil {
		log.L(ctx).Errorf("Failed to load csv to bigquery", err)

		return err
	}

	status, err := job.Wait(ctx)
	if err != nil {
		log.L(ctx).Errorf("Failed to load csv to bigquery", err)

		return err
	}

	if err := status.Err(); err != nil {
		log.L(ctx).Errorf("jobid:%s.Failed to load csv to bigquery:%v", job.ID(), err)

		return err
	}

	log.L(ctx).Infof("load csv to bigquery success")

	return nil
}

func uploadGcs(ctx context.Context, rows []*model.OrgBq) error {
	// Initialize the GCS client
	client, err := storage.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("storage.NewClient: %v", err)
	}
	defer client.Close()

	// Get a handle to the bucket and object (file)
	bucket := client.Bucket(bucketName)
	obj := bucket.Object(fileName)

	// Open a new writer for the GCS object
	wc := obj.NewWriter(ctx)
	// It's recommended to set the "Content-Type" of the file written to GCS
	wc.ContentType = "text/csv"

	// CSV writer writes records to w (io.Writer implemented by wc)
	csvWriter := csv.NewWriter(wc)
	defer csvWriter.Flush() // Don't forget to flush the buffer!

	var headers []string
	val := reflect.ValueOf(rows[0])
	for i := 0; i < val.Type().NumField(); i++ {
		headers = append(headers, val.Type().Field(i).Name)
	}
	err = csvWriter.Write(headers)
	if err != nil {
		wc.Close()
		log.L(ctx).Errorf("csvWriter.Write: %v", err)

		return err
	}
	// Iterate over the rows to write them as CSV records
	for _, row := range rows {
		var record []string
		val := reflect.ValueOf(row)
		for i := 0; i < val.NumField(); i++ {
			field := val.Field(i)
			var str string
			switch field.Kind() {
			case reflect.String:
				str = field.String()
			case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
				str = strconv.FormatInt(field.Int(), 10)
			case reflect.Float32, reflect.Float64:
				str = strconv.FormatFloat(field.Float(), 'f', -1, 64)
			case reflect.Bool:
				str = strconv.FormatBool(field.Bool())
			default:
				str = fmt.Sprintf("%v", field.Interface())
			}
			record = append(record, str)
		}
		err = csvWriter.Write(record)
		if err != nil {
			log.L(ctx).Errorf("csvWriter.Write: %v", err)
			wc.Close()

			return err
		}
	}

	// Check if there were any errors during flush
	if err := csvWriter.Error(); err != nil {
		wc.Close()
		return fmt.Errorf("csvWriter.Error: %v", err)
	}

	// Close the Writer when done to ensure the data is written to GCS
	if err := wc.Close(); err != nil {
		return fmt.Errorf("Writer.Close: %v", err)
	}

	fmt.Printf("Data successfully uploaded to %s/%s\n", bucketName, fileName)
	return nil
}

func loadBqData(ctx context.Context, bqSrv *bigquery.Client, datasetID, tableID string) error {
	// Specify the GCS URI of the CSV file
	gcsRef := bigquery.NewGCSReference(fmt.Sprintf("gs://%s/%s", bucketName, fileName))
	// Set the format of the source data, e.g., CSV, Avro, ORC, Parquet, etc.
	gcsRef.SourceFormat = bigquery.CSV // Set to bigquery.DataFormat("CSV") for CSV format

	// Specify the Load Job configuration to load the data from GCS and replace the existing data in BigQuery
	loader := bqSrv.Dataset(datasetID).Table(tableID).LoaderFrom(gcsRef)
	// Replace the table if it already exists
	loader.WriteDisposition = bigquery.WriteTruncate

	// Run the loader
	job, err := loader.Run(ctx)
	if err != nil {
		log.L(ctx).Errorf("bigquery.LoadFromGCS.Run: %v", err)

		return err
	}

	// Wait for the job to complete
	status, err := job.Wait(ctx)
	if err != nil {
		log.L(ctx).Errorf("bigquery.Job.Wait: %v", err)

		return err
	}

	// If there was an error during the job execution, it will be returned here
	if err := status.Err(); err != nil {
		log.L(ctx).Errorf("bigquery.JobStatus.Err: %v", err)

		return err
	}

	log.L(ctx).Info("Data successfully loaded from GCS and written to BigQuery")

	return nil
}

func execSql(ctx context.Context, bqSrv *bigquery.Client, sql string) error {
	job, err := bqSrv.Query(sql).Run(ctx)
	if err != nil {
		log.L(ctx).Errorf("Query.Run Error:", err)

		return err
	}
	status, err := job.Wait(ctx)
	if err != nil {
		log.L(ctx).Errorf("Job.Wait Error:", err)

		return err
	}
	if status.Err() != nil {
		log.L(ctx).Errorf("Job Status Error:", status.Err())

		return status.Err()
	}
	log.L(ctx).Infof("Exec sql success: %s", sql)

	return nil
}