FeedDataverseBQ/Migration/GoogleStorageController.cs

using System.Text.RegularExpressions;
using Google.Cloud.BigQuery.V2;
using Microsoft.AspNetCore.Mvc;
using Migration.Services;
using Newtonsoft.Json.Linq;
using Newtonsoft.Json;
using System.Text;


public interface IBigQuery
{
    Task<IActionResult> GetProducts();
}

public class GoogleBigQuery : IBigQuery
{
    private readonly BigQueryClient bigqueryClient;
    private readonly string GoogleDatabase;
    private readonly IHttpClientFactory httpClientFactory;

    public GoogleBigQuery(IConfiguration configuration, IHttpClientFactory httpClientFactory, BigQueryClient bigQueryClient)
    {
        this.bigqueryClient = bigQueryClient;
        GoogleDatabase = configuration.GetValue<string>("BigQuery:DbName") ?? throw new Exception("Nie podano parametru nazwy bazy danych BigQuery w konfiguracji");
        this.httpClientFactory = httpClientFactory;
    }

    public async Task<IActionResult> GetProducts()
    {
        try
        {
            int from = 1, limit = 100;
            string me = string.Empty;
            var httpClient = httpClientFactory.CreateClient(DataverseProvider.HTTP_CLIENT);
            do
            {
                var jsonBody = new JObject
                {
                    {"entity_name", "bon_commodity_index"},
                    {"attributes", new JArray {
                        "bon_main_product_id",
                        "bon_index",
                        "bon_gid",
                        "bon_ean",
                        "bon_product_name",
                        "bon_slug_url",
                        "bon_warehouse_state_central",
                        "bon_vat_rate",
                        "bon_converter",
                        "bon_calculaltion_unit",
                        "bon_marketplace",
                        "bon_price_for"
                    }},
                    {"makeFilter", true},
                    {"conditions", new JArray {
                        new JObject {
                            {"attribute_name", "bon_slug_url"},
                            {"condition_operator", "NotNull"}
                        },
                        new JObject {
                            {"attribute_name", "bon_converter"},
                            {"condition_operator", "NotNull"}
                        }
                    }},
                    {"links", new JArray {
                        new JObject {
                            {"primary_entity_name", "bon_commodity_index"},
                            {"related_entity_name", "bon_main_product"},
                            {"primary_key", "bon_main_product_id"},
                            {"foreign_key", "bon_main_productid"},
                            {"join_operator", "Inner"},
                            {"columns", new JArray {
                                "bon_ec_name",
                                "bon_seo_description"
                            }},
                            {"entity_alias", "bon_main_product"}
                        },
                        new JObject {
                            {"primary_entity_name", "bon_commodity_index"},
                            {"related_entity_name", "bon_purchase_index"},
                            {"primary_key", "bon_commodity_indexid"},
                            {"foreign_key", "bon_commodity_index_id"},
                            {"join_operator", "LeftOuter"},
                            {"columns", new JArray {
                                "bon_ecommerce_status"
                            }},
                            {"entity_alias", "bon_purchase_index"}
                        },
                        new JObject {
                            {"primary_entity_name", "bon_main_product"},
                            {"related_entity_name", "bon_producer_farmer"},
                            {"primary_key", "bon_producer_id"},
                            {"foreign_key", "bon_producer_farmerid"},
                            {"join_operator", "Inner"},
                            {"columns", new JArray {
                                "bon_name"
                            }},
                            {"entity_alias", "bon_producer_farmer"}
                        },
                        new JObject {
                            {"primary_entity_name", "bon_commodity_index"},
                            {"related_entity_name", "bon_public_document"},
                            {"primary_key", "bon_commodity_indexid"},
                            {"foreign_key", "bon_commodity_index_id"},
                            {"join_operator", "Inner"},
                            {"columns", new JArray {
                                "bon_url"
                            }},
                            {"entity_alias", "bon_public_document"},
                            {"conditions", new JArray {
                                new JObject {
                                    {"attribute_name", "bon_order"},
                                    {"attribute_value", "1"},
                                    {"condition_operator", "Equal"}
                                },
                                new JObject {
                                    {"attribute_name", "bon_file_type_id"},
                                    {"attribute_value", "c62765b3-2d0c-ee11-8f6e-6045bd8c9e0f"},
                                    {"condition_operator", "Equal"}
                                }
                            }}
                        },
                        new JObject {
                            {"primary_entity_name", "bon_commodity_index"},
                            {"related_entity_name", "bon_sales_price_list"},
                            {"primary_key", "bon_commodity_indexid"},
                            {"foreign_key", "bon_commodity_index_id"},
                            {"join_operator", "Inner"},
                            {"columns", new JArray {
                                "bon_cash_price"
                            }},
                            {"entity_alias", "bon_sales_price_list"},
                            {"conditions", new JArray {
                                new JObject {
                                    {"attribute_name", "bon_publication_type"},
                                    {"attribute_value", "2"},
                                    {"condition_operator", "Equal"}
                                },
                                new JObject {
                                    {"attribute_name", "statuscode"},
                                    {"attribute_value", new JArray {"1", "180430002"}},
                                    {"condition_operator", "In"}
                                },
                                new JObject {
                                    {"attribute_name", "bon_pricelistid"},
                                    {"attribute_value", "48e17eb1-9495-e811-a835-000d3ab48443"},
                                    {"condition_operator", "Equal"}
                                }
                            }}
                        },
                        new JObject {
                            {"primary_entity_name", "bon_main_product"},
                            {"related_entity_name", "bon_categories"},
                            {"primary_key", "bon_category_id"},
                            {"foreign_key", "bon_categoriesid"},
                            {"join_operator", "LeftOuter"},
                            {"columns", new JArray {
                                "bon_categoriesid",
                                "osa_gmc_id"
                            }},
                            {"entity_alias", "bon_categories"}
                        },
                        new JObject{
                            {"primary_entity_name", "bon_main_product"},
                            {"related_entity_name", "bon_categories"},
                            {"primary_key", "bon_category_id"},
                            {"foreign_key", "bon_categoriesid"},
                            {"join_operator", "LeftOuter"},
                            {"columns", new JArray {
                                "bon_short_name"
                            }},
                            {"entity_alias", "bon_category_path"},
                            {"recurece", true}
                        }
                    }}
                };

                var response = await httpClient.PostAsync($"getdata?enviroment=prod&from={from}&limit={limit}", new StringContent(
                                                                                                                    JsonConvert.SerializeObject(jsonBody, Formatting.None),
                                                                                                                    Encoding.UTF8,
                                                                                                                    "application/json"
                                                                                                                ));

                if ((int)response.StatusCode == 500 && JObject.Parse(await response.Content.ReadAsStringAsync())?["errorCode"]?.Value<int>() == 501)
                    break;

                response.EnsureSuccessStatusCode();

                var content = await response.Content.ReadAsStringAsync() ?? throw new Exception("Brak odpowiedzi z Dataverse");
                var jObject = JsonConvert.DeserializeObject<JObject>(content) ?? throw new Exception("Nieprawidłowy format JSON");

                var resultArray = jObject["result"] as JArray ?? throw new Exception("Brak pola 'result' w odpowiedzi");
                var resultList = resultArray.ToObject<List<Dictionary<string, object>>>() ?? new List<Dictionary<string, object>>();

                List<BigQueryInsertRow> insertRows = resultArray!.ToObject<List<JObject>>()!
                    .ConvertAll(row => ConvertToBigQueryInsertRow(row));

                string upsertQuery = $@"
                MERGE `{GoogleDatabase}.bon_main_product` AS T
                USING (
                    SELECT DISTINCT * FROM UNNEST([
                        {{0}}
                    ])
                ) AS S
                ON T.bon_commodity_indexid = S.bon_commodity_indexid
                WHEN MATCHED THEN
                    UPDATE SET
                        bon_cash_price = S.bon_cash_price,
                        bon_categoriesid = S.bon_categoriesid,
                        transactioncurrencyid = S.transactioncurrencyid,
                        bon_warehouse_state_central = S.bon_warehouse_state_central,
                        bon_ec_name = S.bon_ec_name,
                        bon_ean = S.bon_ean,
                        bon_gid = S.bon_gid,
                        bon_index = S.bon_index,
                        bon_name = S.bon_name,
                        bon_url = S.bon_url,
                        bon_category_path = S.bon_category_path,
                        bon_seo_description = S.bon_seo_description,
                        bon_main_product_id = S.bon_main_product_id,
                        bon_slug_url = S.bon_slug_url,
                        bon_converter = S.bon_converter,
                        bon_vat_rate = S.bon_vat_rate,
                        bon_product_name = S.bon_product_name,
                        osa_gmc_id = S.osa_gmc_id,
                        bon_calculaltion_unit = S.bon_calculaltion_unit,
                        bon_marketplace = S.bon_marketplace,
                        bon_price_for = S.bon_price_for,
                        bon_ecommerce_status = S.bon_ecommerce_status
                WHEN NOT MATCHED THEN
                    INSERT (
                        bon_commodity_indexid,
                        bon_cash_price,
                        bon_categoriesid,
                        transactioncurrencyid,
                        bon_warehouse_state_central,
                        bon_ec_name,
                        bon_ean,
                        bon_gid,
                        bon_index,
                        bon_name,
                        bon_url,
                        bon_category_path,
                        bon_seo_description,
                        bon_main_product_id,
                        bon_slug_url,
                        bon_converter,
                        bon_vat_rate,
                        bon_product_name,
                        osa_gmc_id,
                        bon_calculaltion_unit,
                        bon_marketplace,
                        bon_price_for,
                        bon_ecommerce_status
                    )
                    VALUES (
                        S.bon_commodity_indexid,
                        S.bon_cash_price,
                        S.bon_categoriesid,
                        S.transactioncurrencyid,
                        S.bon_warehouse_state_central,
                        S.bon_ec_name,
                        S.bon_ean,
                        S.bon_gid,
                        S.bon_index,
                        S.bon_name,
                        S.bon_url,
                        S.bon_category_path,
                        S.bon_seo_description,
                        S.bon_main_product_id,
                        S.bon_slug_url,
                        S.bon_converter,
                        S.bon_vat_rate,
                        S.bon_product_name,
                        S.osa_gmc_id,
                        S.bon_calculaltion_unit,
                        S.bon_marketplace,
                        S.bon_price_for,
                        S.bon_ecommerce_status
                    )";

                await PutDataBigQuery(upsertQuery, insertRows);

                from++;
            }
            while (true);
            return new OkObjectResult(new { message = me });
        }
        catch (Exception ex)
        {
            Console.WriteLine($"PutDataBigQuery_bon_commodity_index: {ex.Message}");
            return new BadRequestObjectResult(new { message = ex.Message });
        }
    }

    internal async Task PutDataBigQuery(string upsertQuery, List<BigQueryInsertRow> rows)
    {
        try
        {
            var table = bigqueryClient.GetTable(GoogleDatabase, "bon_main_product");

            string CleanString(object value)
            {
                return value != null
                    ? Regex.Replace(value.ToString() ?? "", @"\t|\n|\r", "")
                    : "";
            }

            var insertRows = rows.Select(row => new BigQueryInsertRow
            {
                { "bon_commodity_indexid", row["bon_commodity_indexid"] },
                { "bon_cash_price", row["bon_cash_price"] },
                { "bon_categoriesid", row["bon_categoriesid"] },
                { "transactioncurrencyid", row["transactioncurrencyid"] },
                { "bon_warehouse_state_central", row["bon_warehouse_state_central"] },
                { "bon_ec_name", row["bon_ec_name"] },
                { "bon_ean", row["bon_ean"] },
                { "bon_gid", row["bon_gid"] },
                { "bon_index", row["bon_index"] },
                { "bon_name", row["bon_name"] },
                { "bon_url", row["bon_url"] },
                { "bon_category_path", row["bon_category_path"] },
                { "bon_seo_description", CleanString(row["bon_seo_description"]) },
                { "bon_main_product_id", row["bon_main_product_id"] },
                { "bon_slug_url", row["bon_slug_url"] },
                { "bon_converter", row["bon_converter"] },
                { "bon_vat_rate", row["bon_vat_rate"] },
                { "bon_product_name", row["bon_product_name"] },
                { "osa_gmc_id", row["osa_gmc_id"] },
                { "bon_calculaltion_unit", row["bon_calculaltion_unit"] },
                { "bon_marketplace", row["bon_marketplace"] },
                { "bon_price_for", row["bon_price_for"] },
                { "bon_ecommerce_status", row["bon_ecommerce_status"] }
            }).ToList();

            var mergeValues = insertRows.Select(row =>
                $"STRUCT('{row["bon_commodity_indexid"]}' as bon_commodity_indexid, " +
                $"cast({row["bon_cash_price"] ?? "NULL"} as numeric) as bon_cash_price, " +
                $"'{row["bon_categoriesid"]}' as bon_categoriesid, " +
                $"'{row["transactioncurrencyid"]}' as transactioncurrencyid, " +
                $"cast({row["bon_warehouse_state_central"] ?? "NULL"} as numeric) as bon_warehouse_state_central, " +
                $"'{row["bon_ec_name"]}' as bon_ec_name, " +
                $"'{row["bon_ean"]}' as bon_ean, " +
                $"{row["bon_gid"] ?? "NULL"} as bon_gid, " +
                $"'{row["bon_index"]}' as bon_index, " +
                $"'{row["bon_name"]}' as bon_name, " +
                $"'{row["bon_url"]}' as bon_url, " +
                $"'{row["bon_category_path"]}' as bon_category_path, " +
                $"'{row["bon_seo_description"]}' as bon_seo_description, " +
                $"'{row["bon_main_product_id"]}' as bon_main_product_id, " +
                $"'{row["bon_slug_url"]}' as bon_slug_url, " +
                $"cast({row["bon_converter"] ?? "NULL"} as numeric) as bon_converter, " +
                $"{row["bon_vat_rate"] ?? "NULL"} as bon_vat_rate, " +
                $"'{row["bon_product_name"]}' as bon_product_name, " +
                $"{row["osa_gmc_id"] ?? "NULL"} as osa_gmc_id, " +
                $"'{row["bon_calculaltion_unit"]}' as bon_calculaltion_unit, " +
                $"ARRAY[{row["bon_marketplace"]}] as bon_marketplace, " +
                $"{row["bon_price_for"] ?? "NULL"} as bon_price_for, " +
                $"{row["bon_ecommerce_status"] ?? "NULL"} as bon_ecommerce_status)");

            var queryOptions = new QueryOptions { UseQueryCache = false };
            await bigqueryClient.ExecuteQueryAsync(string.Format(upsertQuery, string.Join(", ", mergeValues)), null, queryOptions);
        }
        catch (Exception ex)
        {
            Console.WriteLine($"PutDataBigQuery: {ex.Message}");
        }
    }

    static BigQueryInsertRow ConvertToBigQueryInsertRow(JObject item)
    {
        BigQueryInsertRow insertRow = new BigQueryInsertRow();
        insertRow.Add("bon_commodity_indexid", (string?)item["bon_commodity_indexid"]);
        insertRow.Add("bon_cash_price", BigQueryNumeric.FromDecimal((decimal)item["bon_cash_price"]!, LossOfPrecisionHandling.Throw));
        insertRow.Add("bon_categoriesid", (string?)item["bon_categoriesid"]);
        insertRow.Add("transactioncurrencyid", (string?)item["transactioncurrencyid"]);
        insertRow.Add("bon_warehouse_state_central", BigQueryNumeric.FromDecimal((decimal)item["bon_warehouse_state_central"]!, LossOfPrecisionHandling.Truncate));
        insertRow.Add("bon_ec_name", (string?)item["bon_ec_name"]);
        insertRow.Add("bon_ean", (string?)item["bon_ean"]);
        insertRow.Add("bon_gid", (int?)item["bon_gid"]);
        insertRow.Add("bon_index", (string?)item["bon_index"]);
        insertRow.Add("bon_name", (string?)item["bon_name"]);
        insertRow.Add("bon_url", (string?)item["bon_url"]);
        insertRow.Add("bon_category_path", (string?)item["bon_category_path"]);
        insertRow.Add("bon_seo_description", (string?)item["bon_seo_description"]);
        insertRow.Add("bon_main_product_id", (string?)item["bon_main_product_id"]);
        insertRow.Add("bon_slug_url", (string?)item["bon_slug_url"]);
        insertRow.Add("bon_converter", BigQueryNumeric.FromDecimal((decimal)item["bon_converter"]!, LossOfPrecisionHandling.Truncate));
        insertRow.Add("bon_vat_rate", (int?)item["bon_vat_rate"] ?? null);
        insertRow.Add("bon_product_name", (string?)item["bon_product_name"]);
        insertRow.Add("osa_gmc_id", (long?)item["osa_gmc_id"]);
        insertRow.Add("bon_calculaltion_unit", (string?)item["bon_calculaltion_unit"]);
        insertRow.Add("bon_marketplace",
            item["bon_marketplace"] is JArray marketplaceArray
                ? string.Join(", ", marketplaceArray.Select(mp => mp["value"]?.ToString()))
                : null);
        insertRow.Add("bon_price_for",
            item["bon_price_for"] is JObject priceForObject
                ? (int?)priceForObject["id"]
                : null);
        insertRow.Add("bon_ecommerce_status",
            item["bon_ecommerce_status"] is JObject statusForObject
                ? (int?)statusForObject["id"]
                : -1);
        return insertRow;
    }
}