Partsource/PartSource.Automation/Jobs/ProcessWhiFitment.cs

using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using PartSource.Automation.Models.Configuration;
using PartSource.Automation.Models.Enums;
using PartSource.Automation.Services;
using Ratermania.Automation.Interfaces;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.IO;
using System.IO.Compression;
using System.Linq;
using System.Security.Cryptography;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;

namespace PartSource.Automation.Jobs
{
	public class ProcessWhiFitment : IAutomationJob
	{
		private readonly ILogger<ProcessWhiFitment> _logger;
		private readonly WhiSeoService _whiSeoService;
		private readonly FtpConfiguration _ftpConfiguration;
		private readonly SeoDataType _seoDataType;

		private readonly IDictionary<string, string> _noteDictionary;

		public ProcessWhiFitment(IConfiguration configuration, ILogger<ProcessWhiFitment> logger, WhiSeoService whiSeoService)
		{
			_logger = logger;
			_whiSeoService = whiSeoService;

			_seoDataType = SeoDataType.Fitment;

			_ftpConfiguration = configuration.GetSection("ftpServers:WhiConfiguration").Get<FtpConfiguration>();

			_noteDictionary = new ConcurrentDictionary<string, string>();
		}

		[System.Diagnostics.CodeAnalysis.SuppressMessage("Reliability", "CA2008:Do not create tasks without passing a TaskScheduler", Justification = "<Pending>")]
		public async Task Run(CancellationToken token, params string[] arguments)
		{
            _whiSeoService.TruncateFitmentTables();
            _whiSeoService.GetFiles(_seoDataType);

            string directory = Path.Combine(_ftpConfiguration.Destination, _seoDataType.ToString().ToLowerInvariant());
			DirectoryInfo directoryInfo = new DirectoryInfo(directory);

			ConcurrentQueue<IGrouping<string, FileInfo>> fileGroups = new ConcurrentQueue<IGrouping<string, FileInfo>>();

			foreach (IGrouping<string, FileInfo> fileGroup in directoryInfo.GetFiles().Where(f => f.Name.EndsWith("csv.gz")).GroupBy(x => x.Name.Split('_').Last()))
			{
				fileGroups.Enqueue(fileGroup);
			}

			Task[] taskArray = new Task[8];

			for (int i = 0; i < taskArray.Length; i++)
			{
				taskArray[i] = Task.Factory.StartNew(() =>
				{
					while (fileGroups.TryDequeue(out IGrouping<string, FileInfo> fileGroup))
					{
						foreach (FileInfo fileInfo in fileGroup)
						{
							try
							{
								string filename = Decompress(fileInfo);
								string tableName = fileInfo.Name.Substring(0, fileInfo.Name.IndexOf('.'));

								DataTable dataTable = GetDataTable(filename);

								_whiSeoService.BulkCopyFitment(dataTable, tableName);
								_logger.LogInformation($"Copied {fileInfo.Name} to the database.");

								File.Delete(filename);
							}

							catch (Exception ex)
							{
								_logger.LogError($"Failed to write {fileInfo.Name} to the database - {ex.Message}", ex);
							}
						}

						string fitmentTable = fileGroup.Key.Substring(0, fileGroup.Key.IndexOf('.'));
						_whiSeoService.CreateFitmentTable(fitmentTable);

						_logger.LogInformation($"Created fitment table for part group {fitmentTable}.");

					}
				});
			}

			Task.WaitAll(taskArray);

			_whiSeoService.SaveNotes(_noteDictionary);
			//_whiSeoService.CreateFitmentView();
		}

		public string Decompress(FileInfo fileInfo)
		{
			string decompressedFile = fileInfo.FullName.Remove(fileInfo.FullName.Length - fileInfo.Extension.Length);

			using FileStream filestream = File.Create(decompressedFile);
			using GZipStream decompressionStream = new GZipStream(fileInfo.OpenRead(), CompressionMode.Decompress);

			decompressionStream.CopyTo(filestream);

			return decompressedFile;
		}

		private DataTable GetDataTable(string filename)
		{
			using DataTable dataTable = new DataTable();
			dataTable.Columns.Add("LineCode", typeof(string));
			dataTable.Columns.Add("PartNumber", typeof(string));
			dataTable.Columns.Add("BaseVehicleId", typeof(int));
			dataTable.Columns.Add("EngineConfigId", typeof(int));
			dataTable.Columns.Add("Position", typeof(string));
			dataTable.Columns.Add("FitmentNoteHash", typeof(string));

			using StreamReader reader = new StreamReader(filename);
			string line = reader.ReadLine(); // Burn the header row

			while (reader.Peek() > 0)
			{
				line = reader.ReadLine();

				string[] columns = line.Split("\",\"");
				for (int i = 0; i < columns.Length; i++)
				{
					columns[i] = columns[i].Replace("\"", string.Empty);
				}

				string lineCode = Regex.Replace(columns[0], "[^a-zA-Z0-9]", string.Empty).Trim();
				string partNumber = Regex.Replace(columns[1], "[^a-zA-Z0-9]", string.Empty).Trim();
				string position = columns[7].Trim();

				string noteText = columns[4].Trim();
				string noteTextHash = GetMD5Hash(noteText);

				if (!_noteDictionary.ContainsKey(noteTextHash))
				{
					_noteDictionary.Add(noteTextHash, noteText);
				}

				if (!string.IsNullOrEmpty(lineCode)
					&& !string.IsNullOrEmpty(partNumber)
					&& int.TryParse(columns[5], out int baseVehicleId)
					&& int.TryParse(columns[6], out int engineConfigId))
				{
					dataTable.Rows.Add(new object[] { lineCode, partNumber, baseVehicleId, engineConfigId, position, noteTextHash });
				}
			}

			return dataTable;
		}

		[System.Diagnostics.CodeAnalysis.SuppressMessage("Security", "CA5351:Do Not Use Broken Cryptographic Algorithms", Justification = "Not used for security")]
		private string GetMD5Hash(string input)
		{
			using MD5 md5 = MD5.Create();

			byte[] inputBytes = Encoding.UTF8.GetBytes(input);
			byte[] hashBytes = md5.ComputeHash(inputBytes);

			StringBuilder stringBuilder = new StringBuilder();

			for (int i = 0; i < hashBytes.Length; i++)
			{
				stringBuilder.Append(hashBytes[i].ToString("X2"));
			}

			return stringBuilder.ToString();
		}
	}
}