460 lines
19 KiB
C#
460 lines
19 KiB
C#
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(8, 4);
|
|
var host = Host.CreateApplicationBuilder(args);
|
|
var commandOptions = host.Configuration.GetSection("CmdOptions").Get<CommandOptions>() ?? new CommandOptions();
|
|
Console.WriteLine($"InputDir:{commandOptions?.InputDir}");
|
|
|
|
var oldestTime = DateTime.ParseExact(commandOptions.OldestTime, "yyyyMM", System.Globalization.DateTimeFormatInfo.InvariantInfo);
|
|
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 =>
|
|
{
|
|
if (commandOptions.IsMock) return;
|
|
options.DatabaseFilter = record => "cferp_test";
|
|
|
|
options.TransformBinary = field => commandOptions != null && commandOptions.Isutf8mb4 ? $"_utf8mb4 0x{field}" : $"0x{field}";
|
|
var noFilterTables = commandOptions.NoFilterTables.Split(",");
|
|
//数据过滤
|
|
options.RecordFilter = async (record, cache) =>
|
|
{
|
|
|
|
if (noFilterTables.Contains(record.TableName)) return true;
|
|
|
|
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)
|
|
{
|
|
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 string orderProcessID))
|
|
{
|
|
var value = await cache.GetStringAsync($"order_process_{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 string id))
|
|
{
|
|
var value = await cache.GetStringAsync($"order_block_plan_{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($"order_block_plan_{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($"order_{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))
|
|
{
|
|
var shardKey =await cache.GetStringAsync($"order_process_{processID}");
|
|
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);
|
|
}
|
|
}
|
|
}
|
|
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.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,
|
|
SslMode = MySqlSslMode.None,
|
|
}.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>();
|
|
if(commandOptions.IsMock)host.Services.AddSingleton<IInputService,SimulationInputService>();
|
|
else 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();
|
|
} |