MES-ETL/MesETL.Test/DataFix.cs

using System.Data;
using MesETL.App.Helpers;
using MesETL.Shared.Helper;
using MySqlConnector;
using Xunit.Abstractions;

namespace TestProject1;

public class DataFix
{
    public const string ConnStr = "Server=192.168.1.245;Port=3306;UserId=root;Password=ruixinjie!@#123;";
    public static string[] DbNames = ["cferp_test_1", "cferp_test_2", "cferp_test_3"];
    
    private readonly ITestOutputHelper _output;
    
    public DataFix(ITestOutputHelper output)
    {
        _output = output;
    }
    
    private string MakeConnStr(string dbName) => ConnStr + $"Database={dbName};";

    /// <summary>
    /// 查找cferp_test_1中CompanyID = 0的order_box_block，根据OrderNo查找对应订单的CompanyID，然后删除重新插入相应的数据库
    /// 如果没有找到对应订单，则保留CompanyID为0
    /// </summary>
    [Fact]
    public async Task FixOrderBoxBlockCompanyID()
    {
        var ds = await DatabaseHelper.QueryTableAsync(MakeConnStr(DbNames[0]), "SELECT * FROM `order_box_block` WHERE COMPANYID = 0");
        var dict = new Dictionary<long, int>();//orderNo -> CompanyID
        foreach (DataRow row in ds.Tables[0].Rows)
        {
            // foreach (var column in row.ItemArray)
            // {
            //     Console.Write(column.ToString() + '\t');
            // }

            var orderNo = Convert.ToInt64(row["OrderNo"]);
            var boxId = Convert.ToInt64(row["BoxID"]);

            int? companyId = null;
            foreach (var db in DbNames)
            {
                if(!dict.TryGetValue(orderNo, out var cid)) // 可以提到外面
                {
                    var result = await DatabaseHelper.QueryScalarAsync(ConnStr + $"Database={db};",
                        $"SELECT CompanyID FROM `order` WHERE OrderNo = {orderNo}");
                    if (result is null or 0) continue;
                    
                    companyId = Convert.ToInt32(result);
                    dict.Add(orderNo, companyId.Value);
                    break;
                }
                else
                {
                    companyId = cid;
                }
            }

            if (companyId is null or 0)
            {
                _output.WriteLine($"OrderBoxBlock:{boxId} - OrderNo {orderNo} not found");
                continue;
            }

            row["CompanyID"] = companyId;
            await DatabaseHelper.TransactionAsync(ConnStr,
                $"""
                 DELETE FROM cferp_test_1.`order_box_block` WHERE BoxID = {boxId};
                 INSERT INTO {TenantDbHelper.GetDbNameByTenantKeyValue(companyId.Value)}.`order_box_block`
                 VALUES(@c1, @c2, @c3, @c4, @c5);
                 """, [new MySqlParameter("c1", row[0]),
                    new MySqlParameter("c2", row[1]),
                    new MySqlParameter("c3", row[2]),
                    new MySqlParameter("c4", row[3]),
                    new MySqlParameter("c5", row[4])]);
            _output.WriteLine($"OrderBoxBock:{boxId} CompanyID -> {companyId}");
        }
    }

    [Theory]
    [InlineData(["order_data_block", "ID", 0])]
    [InlineData(["order_data_block", "ID", 1])]
    [InlineData(["order_data_block", "ID", 2])]
    public async Task FixCompanyIdWithOwnOrderNo(string tableName, string keyName, int dbNameIndex)
    {
        var ds = await DatabaseHelper.QueryTableAsync(MakeConnStr(DbNames[dbNameIndex]), 
            $"SELECT * FROM `{tableName}` WHERE COMPANYID = 0");
        var dict = new Dictionary<long, int>();//orderNo -> CompanyID
        foreach (DataRow row in ds.Tables[0].Rows)
        {
            // foreach (var column in row.ItemArray)
            // {
            //     Console.Write(column.ToString() + '\t');
            // }

            var orderNo = Convert.ToInt64(row["OrderNo"]);
            var key = Convert.ToInt32(row[keyName]);

            int? companyId = null;
            if(!dict.TryGetValue(orderNo, out var cid))
            {
                companyId = await MesDatabaseHelper.TrySearchCompanyId(ConnStr, DbNames,
                    $"SELECT CompanyID FROM `order` WHERE OrderNo = {orderNo}");
            }
            else
            {
                companyId = cid;
            }

            if (companyId is null or 0)
            {
                _output.WriteLine($"{tableName}:{key} - OrderNo {orderNo} not found");
                continue;
            }

            row["CompanyID"] = companyId;
            // ON DUPLICATE KEY UPDATE
            await DatabaseHelper.TransactionAsync(ConnStr,
                $"""
                 DELETE FROM {DbNames[dbNameIndex]}.`{tableName}` WHERE ID = {key};
                 INSERT INTO {TenantDbHelper.GetDbNameByTenantKeyValue(companyId.Value)}.`{tableName}`
                 VALUES({string.Join(',', Enumerable.Range(0, row.ItemArray.Length).Select(i => $"@c{i}"))})
                 ON DUPLICATE KEY UPDATE CompanyID = {companyId};
                 """, 
                Enumerable.Range(0, row.ItemArray.Length).Select(i => new MySqlParameter($"c{i}", row[i])).ToArray());
            _output.WriteLine($"{tableName}:{key} CompanyID -> {companyId}");
        }
    }

    [Theory]
    [InlineData("order", 0)]
    [InlineData("order", 1)]
    [InlineData("order", 2)]
    public async Task FixShardKeyWithOwnOrderNo(string tableName, int dbNameIndex)
    {
        var r = await DatabaseHelper.NonQueryAsync(MakeConnStr(DbNames[dbNameIndex]),
            $"UPDATE `{tableName}` SET ShardKey = CONCAT(SUBSTR(`order`.OrderNo,3,4),'0') WHERE ShardKey = 0");
        _output.WriteLine($"Affect Rows: {r}");
    }
}