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.Text;
using System.Text.RegularExpressions;
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;

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

			_seoDataType = SeoDataType.Fitment;

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

		public async Task Run()
		{
			_whiSeoService.Truncate();
			_whiSeoService.GetFiles(_seoDataType);

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

			ConcurrentQueue<FileInfo> files = new ConcurrentQueue<FileInfo>(directoryInfo.GetFiles().Where(f => f.Name.EndsWith("csv.gz")).OrderBy(f => f.Length));

			while (files.Count > 0)
			{
				Parallel.For(0, 4, index =>
				{
					if (!files.TryDequeue(out FileInfo fileInfo))
					{
						return;
					}

					string filename = Decompress(fileInfo);

					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("NoteText", typeof(string));

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

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

							string[] columns = line.Split("\",\"");

							if (columns.Length != 8)
							{
								skippedLines++;
								continue;
							}

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

							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, noteText });
							}
						}

						string tableName = fileInfo.Name.Substring(0, fileInfo.Name.IndexOf('.'));
						
						_whiSeoService.BulkCopy(_seoDataType, dataTable, tableName);

						if (skippedLines == 0)
						{
							_logger.LogInformation($"Copied {filename} to the database.");
						}

						else
						{
							_logger.LogWarning($"Copied {filename} to the database with warnings.  {skippedLines} lines contained errors and could not be processed.");
						}

						File.Delete(fileInfo.FullName);
					}

					catch (Exception ex)
					{
						_logger.LogError($"Failed to copy {filename} to the database.", ex);
					}

					try
					{
						reader.Close();
						File.Delete(filename);
					}

					catch (Exception ex) { }
				});
			}

			_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;
		}
	}
}