MES-ETL/ConsoleApp2/Program.cs

using ConsoleApp2;
using ConsoleApp2.Const;
using ConsoleApp2.HostedServices;
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using Microsoft.Extensions.Caching.StackExchangeRedis;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using MySqlConnector;
using Serilog;
using Microsoft.Extensions.Caching.Distributed;


await RunProgram();
return;

async Task RunProgram()
{


    //var inputDir= "D:\\MyDumper";
    //ValidateConsole.ValidateInput<string>((_inputDir) =>
    //{
    //    if (Directory.Exists(_inputDir))
    //     {
    //        inputDir = _inputDir;
    //        return true;
    //    }
    //    else return false;
    //}, "请输入读取csv文件的目录(默认为当前目录下MyDumper文件夹):");

    //var maxTask = 16;
    //ValidateConsole.ValidateInput<string>((_inputDir) =>
    //{

    //    _ = int.TryParse(_inputDir.ToString(), out var _taskCount);
    //    if (_taskCount > 0) {
    //        maxTask = _taskCount;
    //        return true;
    //    }
    //    else return false;
    //}, "请输入执行输出的线程数量(默认为16):");

    //var flushCount = 2_0000;
    //ValidateConsole.ValidateInput<string>((_inputDir) =>
    //{
    //    _ = int.TryParse(_inputDir.ToString(), out var _flashCount);
    //    if (_flashCount > 0)
    //    {
    //        flushCount = _flashCount;
    //        return true;
    //    } else return false; 

    //}, "请输入单次插入的行数(默认为20000):");

    ThreadPool.SetMaxThreads(200, 200);
    var host = Host.CreateApplicationBuilder(args);
    var commandOptions = host.Configuration.GetSection("CmdOptions").Get<CommandOptions>() ?? new CommandOptions();
    Console.WriteLine($"InputDir:{commandOptions?.InputDir}");
    Console.WriteLine($"OutPutFlushCount:{commandOptions?.FlushCount}");
    Console.WriteLine($"OutPutTaskCount:{commandOptions?.TaskCount}");

    var oldestTime = DateTime.ParseExact(commandOptions.OldestTime, "yyyyMM", System.Globalization.DateTimeFormatInfo.InvariantInfo);
    host.Services.Configure<InputTableOptions>(option =>
    {
        option.TableInfoConfig = new Dictionary<string, TableInfo>
        {

            //order_block_plan_item从order_item表查询，然后程序插入                
            //order_package_item从order_item表查询，然后程序插入
            //order_patch_detail生产没有这个表，不处理


            {"machine",new TableInfo{SimulaRowCount=14655 }},
            {"order",new TableInfo{SimulaRowCount=5019216 }},
            {"order_block_plan",new TableInfo{SimulaRowCount=2725553 }},//CreateTime < 202301的删除
            {"order_block_plan_result",new TableInfo{SimulaRowCount=1174096 }},
            {"order_box_block",new TableInfo{SimulaRowCount=29755672 }},
            {"order_data_block",new TableInfo{SimulaRowCount=731800334 }},
            {"order_data_goods",new TableInfo{SimulaRowCount=25803671 }},
            {"order_data_parts",new TableInfo{SimulaRowCount=468517543 }},
            {"order_item",new TableInfo{SimulaRowCount=1345520079 }},
            {"order_module",new TableInfo{SimulaRowCount=103325385 }},
            {"order_module_extra",new TableInfo{SimulaRowCount=54361321 }},
            {"order_module_item",new TableInfo{SimulaRowCount=69173339 }},
            {"order_package",new TableInfo{SimulaRowCount=16196195 }},
            {"order_process",new TableInfo{SimulaRowCount=3892685 }},//orderNo < 202301的
            {"order_process_step",new TableInfo{SimulaRowCount=8050349 }},//orderNo < 202301的删除
            {"order_process_step_item",new TableInfo{SimulaRowCount=14538058 }},//orderNo < 202301的删除
            {"order_scrap_board",new TableInfo{SimulaRowCount=123998 }},
            {"process_group",new TableInfo{SimulaRowCount=1253 }},
            {"process_info",new TableInfo{SimulaRowCount=7839 }},
            {"process_item_exp",new TableInfo{SimulaRowCount=28 }},
            {"process_schdule_capacity",new TableInfo{SimulaRowCount=39736 }},
            {"process_step_efficiency",new TableInfo{SimulaRowCount=8 }},
            {"report_template",new TableInfo{SimulaRowCount=7337 }},
            {"simple_package",new TableInfo{SimulaRowCount=130436 }},//orderNo < 202301的删除
            {"simple_plan_order",new TableInfo{SimulaRowCount=351470 }},//CreateTime < 202301的删除
            {"sys_config",new TableInfo{SimulaRowCount=2296 }},
            {"work_calendar",new TableInfo{SimulaRowCount=11 }},
            {"work_shift",new TableInfo{SimulaRowCount=59 }},
            {"work_time",new TableInfo{SimulaRowCount=62 }},
        };
    });
    host.Services.Configure<CsvOptions>(option =>
    {
        option.Delimiter = ",";
        option.QuoteChar = '"';
    });
    host.Services.Configure<DataInputOptions>(options =>
    {
        options.InputDir = commandOptions.InputDir;
        var _csvOptions = new CsvOptions { Delimiter = ",", QuoteChar = '"' };
        options.CreateSource = (string tableName) =>
        {
            var source = new ZstSource(commandOptions.InputDir, tableName, _csvOptions.Delimiter, _csvOptions.QuoteChar);
            return source;
        };
    });

    host.Services.Configure<DataTransformOptions>(options =>
    {
        options.DatabaseFilter = record => "cferp_test";

        options.TransformBinary = field => commandOptions != null && commandOptions.Isutf8mb4 ? $"_utf8mb4 0x{field}" : $"0x{field}";
        //数据过滤
        options.RecordFilter = async (record, cache) =>
        {
            //var index = Array.IndexOf(record.Headers, "ShardKey");
            if (record.TryGetField("ShardKey", out var skStr))
            {
                short.TryParse(skStr, out var sk);
                if (sk < commandOptions.OldestShardKey) return false;
            }

            if (record.TryGetField("CreateTime", out var createTime))
            {
                _ = DateTime.TryParse(createTime.Replace("\"", ""), out var time);
                if (time < oldestTime) return false;
            }
            if (record.TryGetField("OrderNo", out var orderNo))
            {
                try
                {
                    var yearMonth = orderNo.Substring(0, 6);

                    var dt = DateTime.ParseExact(yearMonth, "yyyyMM", System.Globalization.DateTimeFormatInfo.InvariantInfo);
                    if (dt < oldestTime) return false;

                }
                catch (Exception ex)
                {
                    return false;//订单号转换失败，跳过

                }
                
            }

            if (record.TableName == "order_package")
            {
                if (record.TryGetField("PakageNo", out var pkNo))
                {
                    if (pkNo.Length <= 2) return false;
                }
            }
            if (record.TableName == "order_block_plan")
            {
                if (record.TryGetField("OrderNos", out var nos))
                {
                    if (nos.Length <= 2) return false;
                }
              
            }
            if (record.TableName == "order_process_step" || record.TableName == "order_process_step_item")
            {
                //如果缓存中不存在OrderProcessID,则丢弃
                if(record.TryGetField("OrderProcessID",out var orderProcessID))
                {
                    var value = await cache.GetStringAsync(orderProcessID);
                    if (string.IsNullOrEmpty(value))return false;
                }
            }
            if (record.TableName == "order_block_plan_result" )
            {
                //如果缓存中不存在ID,则丢弃(ID 对应order_block_plan中的ID)
                if (record.TryGetField("ID", out var id))
                {
                    var value = await cache.GetStringAsync(id);
                    if (string.IsNullOrEmpty(value)) return false;
                }
            }
            return true;

        };
        //数据修改
        options.RecordModify = (record) =>
        {
            if (record.TableName == "order_block_plan")
            {
                if (record.TryGetField("OrderNos", out var nos))
                {
                    if (nos.Length <= 2) record.SetField("OrderNos", "\"[]\"");
                }

            }
            if (record.TableName == "order_process")//修改order_process.NextStepID的默认值为0
            {
                 
                if (record.TryGetField("NextStepID", out var idStr))
                {

                    if (idStr == "\\N")
                    {
                        record.SetField("NextStepID", "0");
                    }
                }
            }

        };
        //数据缓存
        options.RecordCache = async (record, cache) =>
        {
            if (record.TableName == "order")
            {
                if (record.TryGetField("OrderNo", out var orderNo))
                {
                    if (record.TryGetField("CompanyID", out var companyid))
                    {
                        await cache.SetStringAsync(orderNo, companyid);
                    }
                    
                }
            }
            if (record.TableName == "order_process")
            {
                if (record.TryGetField("OrderNo", out var orderNo))
                {
                    var yearMonth = orderNo.Substring(2, 4);
                    var sk = yearMonth + "0";
                   
                    if( record.TryGetField("ID", out var id))
                    {
                        await cache.SetStringAsync(id, sk);
                    }
          
                }
            }
            if (record.TableName == "order_block_plan")
            {
                if (record.TryGetField("CompanyID", out var companyid))
                {
                    record.TryGetField("ID", out var id);
                    await cache.SetStringAsync(id, companyid);
                }
            }
        };
        //数据替换
        options.RecordReplace = async (record, cache) =>
        {
            //删除数据源里simple_plan_order.ProcessState 字段和值

            if (record.TableName == "simple_plan_order")//修改order_process.NextStepID的默认值为0
            {
                var nextStepIdIndex = Array.IndexOf(record.Headers, "ProcessState");
                if (nextStepIdIndex > -1)
                {
                    var headers = record.Headers.Where(t => t != "ProcessState").ToArray();
                    var fs = record.Fields.ToList();
                    fs.RemoveAt(nextStepIdIndex);
                    var fields = fs.ToArray();
                    return new DataRecord(fields, record.TableName, headers, record.CompanyID);
                }
            }
            if (record.TableName == "order")//修改order_process.NextStepID的默认值为0
            {
                var nextStepIdIndex = Array.IndexOf(record.Headers, "IsBatch");
                if (nextStepIdIndex > -1)
                {
                    var headers = record.Headers.Where(t => t != "IsBatch").ToArray();
                    var fs = record.Fields.ToList();
                    fs.RemoveAt(nextStepIdIndex);
                    var fields = fs.ToArray();
                    return new DataRecord(fields, record.TableName, headers, record.CompanyID);
                }
            }
            if (record.TableName == "order_block_plan_result")//修改order_process.NextStepID的默认值为0
            {
                if (record.TryGetField("ID", out var id))
                {
                    var headers = new List<string>(record.Headers);
                    var fields =new List<string>(record.Fields);
                    headers.Add("CompanyID");
                    var companyidResult =await cache.GetStringAsync(id);
                    _ = int.TryParse(companyidResult.ToString(), out var companyid);
                    fields.Add(companyid.ToString());
                    return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), companyid);
                }
            }
            if(record.TableName == "order_box_block")
            {
                if (!record.TryGetField("CompanyID", out var companyid))
                {
                    if (record.TryGetField("OrderNo", out var orderNo))
                    {
                        var headers = new List<string>(record.Headers);
                        var fields = new List<string>(record.Fields);
                        headers.Add("CompanyID");
                        var companyidResult = await cache.GetStringAsync(orderNo);
                        _ = int.TryParse(companyidResult, out var cpid);
                        fields.Add(cpid.ToString());
                        return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), cpid);
                    }
                }
            }
            if (record.TableName == "order_module")
            {
                if (record.TryGetField("ViewFileName",out var value))
                {
                    var index=Array.IndexOf(record.Headers, "ViewFileName");
                    var headers = new List<string>(record.Headers);
                    headers.RemoveAt(index); 
                    var fields = new List<string>(record.Fields);
                    fields.RemoveAt(index);
                    return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), record.CompanyID);
                }

            }
            if (record.TableName == "order_process")
            {
                if (!record.TryGetField("ShardKey", out var skStr))
                {
                    if(record.TryGetField("OrderNo", out var orderNo))
                    {
                        var yearMonth = orderNo.Substring(2, 4);
                        var sk = yearMonth + "0";
                        var headers = new List<string>(record.Headers);
                        var fields = new List<string>(record.Fields);
                        headers.Add("ShardKey");
                        fields.Add(sk);
                        return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), record.CompanyID);
                    }
                }
            }
            if(record.TableName == "order_process_step"|| record.TableName == "order_process_step_item")
            {
                if (!record.TryGetField("ShardKey",out var sk))
                {
                    if (record.TryGetField("OrderProcessID",out var processID))
                    {
                        try
                        {

                        var shardKey =await cache.GetStringAsync(processID);
                        var headers = new List<string>(record.Headers);
                        var fields = new List<string>(record.Fields);
                        headers.Add("ShardKey");
                        fields.Add(shardKey);
                        return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), record.CompanyID);
                        }
                        catch (Exception ex)
                        {

                        }
                    }
                }
            }
            if(record.TableName == "order_moudle")
            {
                if (!record.TryGetField("ShardKey", out var skStr))
                {
                    if (record.TryGetField("OrderNo", out var orderNo))
                    {
                        var yearMonth = orderNo.Substring(2, 4);
                        var sk = yearMonth + "0";
                        var headers = new List<string>(record.Headers);
                        var fields = new List<string>(record.Fields);
                        headers.Add("ShardKey");
                        fields.Add(sk);
                        return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), record.CompanyID);
                    }
                }
            }
            return null;
        };
        //数据生成
        options.RecordAdd = (record) =>
        {
            var resultList = new List<DataRecord>();
            if (record.TableName == "order_item")
            {
                record.TryGetField("ID", out var itemID);
                record.TryGetField("ShardKey", out var shardKey);
                record.TryGetField("PlanID", out var planID);
                record.TryGetField("PackageID", out var packageID);
                record.TryGetField("CompanyID", out var companyID);
                _=int.TryParse(planID, out var pid);
                if (pid > 0)
                {
                    resultList.Add(new DataRecord(new[] { itemID, shardKey, planID, companyID },
                        "order_block_plan_item",
                        new[] { "ItemID", "ShardKey", "PlanID", "CompanyID" }));
                }
                _ = int.TryParse(packageID, out var pkid);
                if(pkid > 0)
                {
                    resultList.Add(new DataRecord(new[] { itemID, shardKey, packageID, companyID },
                        "order_package_item",
                        new[] { "ItemID", "ShardKey", "PackageID", "CompanyID" }
                        ));
                }
            }
            return resultList;

        };
        options.ColumnTypeConfig = new()
        {
            { "simple_plan_order.PlaceData", ColumnType.Blob },
            { "order_block_plan_result.PlaceData", ColumnType.Blob },
            { "order_box_block.Data", ColumnType.Blob },
            { "order_data_goods.ExtraProp", ColumnType.Text },
            { "order_module_extra.JsonStr", ColumnType.Text },
            { "process_info.Users", ColumnType.Text },
            { "order_process_schdule.CustomOrderNo", ColumnType.Text },
            { "order_process_schdule.OrderProcessStepName", ColumnType.Text },
            { "order_process_schdule.AreaName", ColumnType.Text },
            { "order_process_schdule.ConsigneeAddress", ColumnType.Text },
            { "order_process_schdule.ConsigneePhone", ColumnType.Text },
            { "report_source.Sql", ColumnType.Text },
            { "report_source.KeyValue", ColumnType.Text },
            { "report_source.Setting", ColumnType.Text },
            { "order_data_block.RemarkJson", ColumnType.Text },
            { "order_patch_detail.BlockDetail", ColumnType.Text },
            { "order_scrap_board.OutLineJson", ColumnType.Text },
            { "simple_package.Items", ColumnType.Text },
            { "order_batch_pack_config.Setting", ColumnType.Text },
            { "machine.Settings", ColumnType.Text },
            { "sys_config.Value", ColumnType.Text },
            { "sys_config.JsonStr", ColumnType.Text },
            { "process_item_exp.ItemJson", ColumnType.Text },
            { "report_template.Template", ColumnType.Text },
            { "report_template.SourceConfig", ColumnType.Text },
            { "order_block_plan.OrderNos", ColumnType.Text },
            { "order_block_plan.BlockInfo", ColumnType.Text },
        };
    });
    host.Services.Configure<DatabaseOutputOptions>(options =>
    {
        //options.ConnectionString = new MySqlConnectionStringBuilder
        //{
        //    Server = "127.0.0.1",
        //    Port = 33309,
        //    Database = "cferp_test",
        //    UserID = "root",
        //    Password = "123456",
        //    MaximumPoolSize = 50, // 这个值应当小于 max_connections
        //}.ConnectionString;
        options.ConnectionString = new MySqlConnectionStringBuilder(host.Configuration.GetConnectionString("MySqlMaster"))
        {
            CharacterSet = "utf8",
            AllowUserVariables = true,
            IgnoreCommandTransaction = true,
            TreatTinyAsBoolean = false,
            MaximumPoolSize = 50
        }.ConnectionString;
        options.TaskCount = commandOptions.TaskCount;
        options.FlushCount = commandOptions.FlushCount;
    });
    host.Services.AddLogging(builder =>
    {
        builder.ClearProviders();
        builder.AddSerilog(new LoggerConfiguration().WriteTo.Console().CreateLogger());
    });

    host.Services.AddSingleton<ProcessContext>();
    host.Services.AddKeyedSingleton<DataRecordQueue>(ProcessStep.Producer);
    host.Services.AddKeyedSingleton<DataRecordQueue>(ProcessStep.Consumer);
    host.Services.AddTransient<TaskManager>();

    host.Services.AddHostedService<MainHostedService>();
    host.Services.AddHostedService<TaskMonitorService>();
    host.Services.AddSingleton<IInputService, InputService>();
    host.Services.AddSingleton<ITransformService, TransformService>();
    host.Services.AddSingleton<IOutputService, OutputService>();
    var redisOptions = host.Configuration.GetSection("RedisCacheOptions").Get<RedisCacheOptions>() ?? new RedisCacheOptions();
    host.Services.AddStackExchangeRedisCache(options =>
    {
        options.Configuration = redisOptions.Configuration;
        options.InstanceName = redisOptions.InstanceName;
    });
    var app = host.Build();
    await app.RunAsync();
}