· 8 years ago · Jul 24, 2018, 10:48 AM
1using System;
2using System.Collections.Generic;
3using System.Data;
4using System.Data.SqlClient;
5using System.Globalization;
6using System.IO;
7using System.Linq;
8using System.Threading;
9using System.Threading.Tasks;
10using FastMember;
11using Infrastructure.CQRS.Abstractions.Commands;
12using Legacy.Models;
13using MediatR;
14using Microsoft.EntityFrameworkCore;
15using Prices.DataAccess;
16using Prices.Services.Abstractions.Read;
17using ProductPrice = Prices.Services.Abstractions.Read.Models.ProductPrice;
18
19namespace Prices.CQRS.Commands.Prices.ImportPrices
20{
21 public class ImportPricesCommandHandler : ICommandHandler<ImportPricesCommand>
22 {
23 private readonly PricesDbContext _dbContext;
24 private readonly IPriceReaderFactory _priceReaderFactory;
25
26 public ImportPricesCommandHandler(
27 PricesDbContext dbContext,
28 IPriceReaderFactory priceReaderFactory)
29 {
30 _dbContext = dbContext ?? throw new ArgumentNullException(nameof(dbContext));
31 _priceReaderFactory = priceReaderFactory ?? throw new ArgumentNullException(nameof(priceReaderFactory));
32 }
33
34 public async Task<Unit> Handle(
35 ImportPricesCommand command,
36 CancellationToken cancellationToken)
37 {
38 cancellationToken.ThrowIfCancellationRequested();
39 var warehouses = await _dbContext
40 .Warehouses
41 .Include(x => x.PriceListConfiguration)
42 .Where(x => x.PriceListConfiguration != null)
43 .AsNoTracking()
44 .ToArrayAsync(cancellationToken);
45
46 foreach (var warehouse in warehouses)
47 {
48 var warehouseId = warehouse.Id;
49 var configuration = warehouse.PriceListConfiguration;
50 var warehouseDirectory = GetPricesDirectoryIfExists(command.PricesDirectory, configuration.FolderName);
51 if (warehouseDirectory == null)
52 continue;
53 var options = new PriceOptions(
54 configuration.ArticleRowIndex,
55 configuration.MakerRowIndex,
56 configuration.TitleRowIndex,
57 configuration.QuantityRowIndex,
58 configuration.PriceRowIndex);
59 var pricesToRead = GetPricesToRead(warehouseDirectory);
60
61 foreach (var priceToRead in pricesToRead)
62 {
63 var errorOccured = false;
64 try
65 {
66 var fileCreationDate = priceToRead.CreationTime;
67 var reader = _priceReaderFactory.CreateReader(priceToRead, options);
68 var prices = reader.ReadPrices();
69
70 using (var connection = GetDetachedConnection())
71 {
72 await connection.OpenAsync(cancellationToken);
73 using (var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted))
74 {
75 await InsertStockInfoIntosTempTableAsync(
76 prices,
77 fileCreationDate,
78 warehouseId,
79 transaction,
80 cancellationToken);
81 transaction.Commit();
82 }
83 }
84 }
85 catch (OperationCanceledException)
86 {
87 throw;
88 }
89 catch (Exception)
90 {
91 errorOccured = true;
92 }
93
94 MoveFile(
95 priceToRead,
96 errorOccured
97 ? command.ErrorDirectory
98 : command.SuccessfulDirectory);
99 }
100 }
101
102 return Unit.Value;
103 }
104
105 private DirectoryInfo GetPricesDirectoryIfExists(
106 DirectoryInfo baseDirectory,
107 string directoryNameInsideBaseDirectory)
108 {
109 var fullPath = Path.Combine(baseDirectory.FullName, directoryNameInsideBaseDirectory);
110 var directory = new DirectoryInfo(fullPath);
111 if (directory.Exists) return directory;
112
113 return null;
114 }
115
116 private FileInfo[] GetPricesToRead(DirectoryInfo directory)
117 {
118 var files = "*.xls|*.xlsx|*.csv|*.zip"
119 .Split(new[] {'|'}, StringSplitOptions.RemoveEmptyEntries)
120 .Select(x => directory.GetFiles(x, SearchOption.TopDirectoryOnly))
121 .SelectMany(x => x)
122 .Distinct()
123 .ToArray();
124 return files;
125 }
126
127 private SqlConnection GetDetachedConnection()
128 {
129 var connection = _dbContext.Database.GetDbConnection();
130 var newConnection = new SqlConnection(connection.ConnectionString);
131
132 return newConnection;
133 }
134
135 private async Task InsertStockInfoIntosTempTableAsync(
136 IEnumerable<ProductPrice> prices,
137 DateTime fileCreationDate,
138 int warehouseId,
139 SqlTransaction transaction,
140 CancellationToken cancellationToken)
141 {
142 cancellationToken.ThrowIfCancellationRequested();
143 var stockInfos = prices.Select(x => new StockInfo
144 {
145 Article = x.VendorCode,
146 Title = x.Title,
147 Maker = x.Manufacturer,
148 Price = x.Price,
149 Quantity = Convert.ToDouble(x.Quantity),
150 ActualDate = fileCreationDate,
151 StorehouseId = warehouseId
152 });
153
154 await CreateTempTableAsync("#tempStockInfo", transaction, cancellationToken);
155 await CopyStockInfoIntoTempTableAsync("#tempStockInfo", stockInfos, transaction, cancellationToken);
156 await RemoveDuplicatesAsync("#tempStockInfo", transaction, cancellationToken);
157 await MergeTempTableIntoMainAsync("#tempStockInfo", warehouseId, transaction, cancellationToken);
158 }
159
160 private async Task CreateTempTableAsync(
161 string tempTableName,
162 SqlTransaction transaction,
163 CancellationToken cancellationToken)
164 {
165 cancellationToken.ThrowIfCancellationRequested();
166 using (var command = transaction.Connection.CreateCommand())
167 {
168 command.Transaction = transaction;
169 command.CommandType = CommandType.Text;
170 command.CommandText = $@"
171DROP TABLE IF EXISTS {tempTableName};
172CREATE TABLE {tempTableName}
173(
174 [Article] [nvarchar](255) NOT NULL,
175 [Title] [nvarchar](255) NOT NULL,
176 [Maker] [nvarchar](255) NOT NULL,
177 [Price] [decimal](18, 2) NOT NULL,
178 [Quantity] [float] NOT NULL,
179 [ActualDate] [datetime2](7) NOT NULL,
180 [StorehouseId] [int] NOT NULL
181);";
182 command.CommandTimeout = 30;
183 await command.ExecuteNonQueryAsync(cancellationToken);
184 }
185 }
186
187 private async Task CopyStockInfoIntoTempTableAsync(
188 string tempTableName,
189 IEnumerable<StockInfo> stockInfos,
190 SqlTransaction transaction,
191 CancellationToken cancellationToken)
192 {
193 cancellationToken.ThrowIfCancellationRequested();
194 using (var bulkCopy = new SqlBulkCopy(transaction.Connection, SqlBulkCopyOptions.Default, transaction))
195 using (var reader = ObjectReader.Create(
196 stockInfos,
197 nameof(StockInfo.Article),
198 nameof(StockInfo.Title),
199 nameof(StockInfo.Maker),
200 nameof(StockInfo.Price),
201 nameof(StockInfo.Quantity),
202 nameof(StockInfo.ActualDate),
203 nameof(StockInfo.StorehouseId)))
204 {
205 bulkCopy.DestinationTableName = tempTableName;
206 bulkCopy.EnableStreaming = true;
207 bulkCopy.BulkCopyTimeout = 3600;
208 await bulkCopy.WriteToServerAsync(reader, cancellationToken);
209 }
210 }
211
212 private async Task RemoveDuplicatesAsync(
213 string tempTableName,
214 SqlTransaction transaction,
215 CancellationToken cancellationToken)
216 {
217 cancellationToken.ThrowIfCancellationRequested();
218 using (var command = transaction.Connection.CreateCommand())
219 {
220 var tempNonDuplicatesTableName = tempTableName + "NonDuplicates";
221 command.Transaction = transaction;
222 command.CommandType = CommandType.Text;
223 command.CommandText = $@"
224DROP TABLE IF EXISTS {tempNonDuplicatesTableName};
225CREATE TABLE {tempNonDuplicatesTableName}
226(
227 [Article] [nvarchar](255) NOT NULL,
228 [Title] [nvarchar](255) NOT NULL,
229 [Maker] [nvarchar](255) NOT NULL,
230 [Price] [decimal](18, 2) NOT NULL,
231 [Quantity] [float] NOT NULL,
232 [ActualDate] [datetime2](7) NOT NULL,
233 [StorehouseId] [int] NOT NULL
234);
235INSERT INTO {tempNonDuplicatesTableName}
236SELECT
237 [Article],
238 [Title],
239 [Maker],
240 [Price],
241 SUM([Quantity]) AS [Quantity],
242 [ActualDate],
243 [StorehouseId]
244FROM {tempTableName}
245GROUP BY [Article], [Title], [Maker], [Price], [ActualDate], [StorehouseId];
246TRUNCATE TABLE {tempTableName};
247INSERT INTO {tempTableName} SELECT [Article], [Title], [Maker], [Price], [Quantity], [ActualDate], [StorehouseId] FROM {tempNonDuplicatesTableName};
248DROP TABLE IF EXISTS {tempNonDuplicatesTableName};";
249 command.CommandTimeout = 3600;
250 await command.ExecuteNonQueryAsync(cancellationToken);
251 }
252 }
253
254 private async Task MergeTempTableIntoMainAsync(
255 string tempTableName,
256 int warehouseId,
257 SqlTransaction transaction,
258 CancellationToken cancellationToken)
259 {
260 cancellationToken.ThrowIfCancellationRequested();
261 using (var command = transaction.Connection.CreateCommand())
262 {
263 command.Transaction = transaction;
264 command.CommandType = CommandType.Text;
265
266 command.CommandText = $@"
267MERGE INTO [dbo.Sales].[StockInfo] as dst
268USING {tempTableName} as src
269ON (dst.[Article] = src.[Article] AND dst.[Title] = src.[Title] AND dst.[Maker] = src.[Maker] AND dst.[Price] = src.[Price] AND dst.[Storehouse_Id] = src.[StorehouseId])
270WHEN MATCHED THEN
271 UPDATE SET dst.[Quantity] = src.[Quantity], dst.[ActualDate] = src.[ActualDate]
272WHEN NOT MATCHED BY TARGET THEN
273 INSERT ([Article], [Title], [Maker], [Price], [Quantity], [ActualDate], [Storehouse_Id])
274 VALUES (src.[Article], src.[Title], src.[Maker], src.[Price], src.[Quantity], src.[ActualDate], src.[StorehouseId])
275WHEN NOT MATCHED BY SOURCE AND dst.[Storehouse_Id] = {warehouseId.ToString("D", CultureInfo.InvariantCulture)} THEN DELETE;";
276 command.CommandTimeout = 3600;
277 await command.ExecuteNonQueryAsync(cancellationToken);
278 }
279 }
280
281 private void MoveFile(FileInfo file, DirectoryInfo destinationDirectory)
282 {
283 var destinationFilePath = Path.Combine(destinationDirectory.FullName, file.Name);
284 var destinationFileInfo = new FileInfo(destinationFilePath);
285 if (destinationFileInfo.Exists)
286 destinationFileInfo.Delete();
287 file.MoveTo(destinationFileInfo.FullName);
288 }
289 }
290}