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;
using Serilog.Events;


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);
    host.Configuration.AddJsonFile(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "appsettings.json"), false, false);
    var commandOptions = host.Configuration.GetSection("CmdOptions").Get<CommandOptions>() ?? 
                         throw new ApplicationException("未发现配置文件");
    Console.WriteLine($"InputDir:{commandOptions?.InputDir}");

    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 =>
    {
        var tenantDbOptions = host.Configuration.GetRequiredSection("TenantDb").Get<TenantDbOptions>() 
            ?? throw new ApplicationException("分库配置项不存在");
        options.DatabaseFilter = record =>
        {
            var companyId = int.Parse(record[tenantDbOptions.TenantKey]); // 每个实体都应存在CompanyID，否则异常
            return tenantDbOptions.GetDbNameByTenantKeyValue(companyId);
        };

        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(record.GetCacheKey("OrderProcessID"));
                    if (string.IsNullOrEmpty(value))return false;
                
            }
            if (record.TableName == "order_block_plan_result" )
            {
                //如果缓存中不存在ID,则丢弃(ID 对应order_block_plan中的ID)

                    var value = await cache.GetStringAsync(record.GetCacheKey("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(record.GetCacheKey("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(record.GetCacheKey("ID"), sk);
                    }
          
                }
            }
            if (record.TableName == "order_block_plan")
            {
                if (record.TryGetField("CompanyID", out var companyid))
                {
                    record.TryGetField("ID", out var id);
                    await cache.SetStringAsync(record.GetCacheKey("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(record.GetCacheKey("ID"));

                    _ = int.TryParse(companyidResult, 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(record.GetCacheKey("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.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")
            {
                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))
                    {
                        var shardKey =await cache.GetStringAsync(record.GetCacheKey("OrderProcessID"));
                        var headers = new List<string>(record.Headers);
                        var fields = new List<string>(record.Fields);
                        headers.Add("ShardKey");
                        fields.Add(shardKey??"0");
                        return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), record.CompanyID);
                    }
                }
            }
            
            // 旧数据库order_data_block.CompanyID数据有误，重新计算
            if (record.TableName == "order_data_block")  
            {
                var companyID = await cache.GetStringAsync($"order_{record["OrderNo"]}");
                if (string.IsNullOrEmpty(companyID))
                    // throw new InvalidDataException($"没有获取到OrderNo={record["OrderNo"]}的CompanyID");
                    return record;
                record["CompanyID"] = companyID;
                return record;
            }
            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.Json },
            { "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;
    });
    
    host.Services.AddLogging(builder =>
    {
        builder.ClearProviders();
        builder.AddSerilog(new LoggerConfiguration()
            .WriteTo.Console()
            .WriteTo.File("./log/error.log", restrictedToMinimumLevel:LogEventLevel.Error)
            // .WriteTo.File("./log/info.log", restrictedToMinimumLevel:LogEventLevel.Information) //性能考虑暂不使用
            .CreateLogger()
        );
    });

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

    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();
}