<commit_before>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;

namespace EtlGate.Core
{
	public interface IDelimitedDataReader
	{
		IEnumerable<Record> ReadFrom(Stream stream, string fieldSeparator = ",", string recordSeparator = "\r\n", bool supportQuotedFields = true, bool hasHeaderRow = false);
	}

	public class DelimitedDataReader : IDelimitedDataReader
	{
		private readonly IStreamTokenizer _streamTokenizer;

		public DelimitedDataReader(IStreamTokenizer streamTokenizer)
		{
			_streamTokenizer = streamTokenizer;
		}

		public IEnumerable<Record> ReadFrom(Stream stream, string fieldSeparator = ",", string recordSeparator = "\r\n", bool supportQuotedFields = true, bool hasHeaderRow = false)
		{
			if (stream == null)
			{
				throw new ArgumentException();
			}

			if (fieldSeparator == null)
			{
				fieldSeparator = "";
			}

			if (recordSeparator == null)
			{
				recordSeparator = "";
			}

			if (fieldSeparator == recordSeparator)
			{
				throw new ParseException("field separator and record separator must be different.");
			}

			var bufferedStream = new BufferedStream(stream, 131072);

			var fieldValues = new List<string>();
			var parseContext = new ParseContext
				                   {
					                   HeaderRowFieldIndexes = new Dictionary<string, int>(),
					                   HasHeaderRow = hasHeaderRow,
					                   RecordSeparator = recordSeparator,
					                   FieldSeparator = fieldSeparator,
					                   IndexOfRecordSeparatorInFieldSeparator = fieldSeparator.IndexOf(recordSeparator, StringComparison.Ordinal),
					                   ReadingHeaderRow = hasHeaderRow,
					                   Handle = StartReadField,
					                   RecordNumber = 1,
					                   SupportQuotedFields = supportQuotedFields,
					                   Capture = new StringBuilder()
				                   };
			var specialTokens = new List<string>();
			if (!String.IsNullOrEmpty(fieldSeparator))
			{
				specialTokens.Add(fieldSeparator);
			}
			if (!String.IsNullOrEmpty(recordSeparator))
			{
				specialTokens.Add(recordSeparator);
			}
			if (supportQuotedFields)
			{
				specialTokens.Add("\"");
			}
			var specialChars = (fieldSeparator + recordSeparator + (supportQuotedFields ? "\"" : "")).Distinct().ToArray();
			foreach (var token in _streamTokenizer.Tokenize(bufferedStream, specialChars))
			{
				var yieldIt = parseContext.Handle(token, fieldValues, parseContext);
				if (yieldIt)
				{
					yield return new Record(fieldValues, parseContext.HeaderRowFieldIndexes);
					fieldValues = parseContext.HasHeaderRow
						              ? new List<string>(parseContext.HeaderRowFieldIndexes.Count)
						              : new List<string>(fieldValues.Count);
					parseContext.Capture.Clear();
				}
			}
			if (parseContext.Handle == ReadQuotedField)
			{
				throw new ParseException("Quoted field does not have a close quote on line " + parseContext.RecordNumber + " field " + parseContext.FieldNumber);
			}
			if (parseContext.ReadingQuotedField && parseContext.Handle == ContinueCollectSeparator)
			{
				ThrowUnescapedQuoteException(parseContext);
			}
			if (parseContext.Capture.Length > 0 ||
			    parseContext.FieldNumber > 0 &&
			    (parseContext.Handle == StartReadField || parseContext.Handle == BranchReadEscapedQuoteOrStartCollectingSeparator))
			{
				if (parseContext.Handle == StartReadField)
				{
					parseContext.IncrementFieldNumber();
				}
				AddFieldValue(fieldValues, parseContext);
			}
			if (fieldValues.Count > 0)
			{
				yield return new Record(fieldValues, parseContext.HeaderRowFieldIndexes);
			}
		}

		private static void AddFieldValue(ICollection<string> fieldValues, ParseContext parseContext)
		{
			var capture = parseContext.Capture;
			var value = capture.ToString();
			fieldValues.Add(value);
			capture.Length = 0;
		}

		private static bool BranchReadEscapedQuoteOrStartCollectingSeparator(Token token, List<string> fieldValues, ParseContext parseContext)
		{
			if (token is DataToken)
			{
				ThrowUnescapedQuoteException(parseContext);
			}

			switch (token.Value)
			{
				case "\"":
					parseContext.Capture.Append('"');
					parseContext.Handle = ReadQuotedField;
					break;
				default:
					if (parseContext.FieldSeparator.Length > 0 && token.Value[0] == parseContext.FieldSeparator[0] ||
					    parseContext.RecordSeparator.Length > 0 && token.Value[0] == parseContext.RecordSeparator[0])
					{
						// possible end of field, more fields follow
						return StartCollectingSeparator(token, fieldValues, parseContext);
					}

					ThrowUnescapedQuoteException(parseContext);
					break;
			}
			return false;
		}

		private static void CheckHeaderRowForDuplicateFieldNames(IEnumerable<string> fieldValues)
		{
			var uniqueHeaderLookup = new HashSet<string>();
			foreach (var field in fieldValues.Where(field => !uniqueHeaderLookup.Add(field)))
			{
				throw new ParseException(String.Format("Header row must not have more than one field with the same name. '{0}' appears more than once in the header row.", field));
			}
		}

		private static bool ContinueCollectSeparator(Token token, List<string> fieldValues, ParseContext parseContext)
		{
			var capture = parseContext.Capture;
			capture.Append(token.Value);
			var potentialSeparator = capture.ToString(capture.Length - parseContext.SeparatorLength, parseContext.SeparatorLength);
			var fieldSeparatorIndex = parseContext.FieldSeparator.Length > 0
				                          ? potentialSeparator.IndexOf(parseContext.FieldSeparator, StringComparison.Ordinal)
				                          : -1;
			var recordSeparatorIndex = parseContext.RecordSeparator.Length > 0
				                           ? potentialSeparator.IndexOf(parseContext.RecordSeparator, StringComparison.Ordinal)
				                           : -1;

			if (fieldSeparatorIndex != -1 && recordSeparatorIndex != -1)
			{
				return ReadFirstSeparator(fieldValues, parseContext, fieldSeparatorIndex, recordSeparatorIndex, potentialSeparator);
			}

			if (fieldSeparatorIndex != -1)
			{
				return TryReadFieldSeparator(fieldValues, parseContext, fieldSeparatorIndex, potentialSeparator);
			}

			if (recordSeparatorIndex != -1)
			{
				if (parseContext.IndexOfRecordSeparatorInFieldSeparator <= 0)
				{
					return TryReadRecordSeparator(fieldValues, parseContext, recordSeparatorIndex, potentialSeparator);
				}

				// implicit:	parseContext.IndexOfRecordSeparatorInFieldSeparator > 0
				var potentialFieldSeparator = potentialSeparator.Substring(recordSeparatorIndex - parseContext.IndexOfRecordSeparatorInFieldSeparator);
				if (potentialFieldSeparator.Length >= parseContext.FieldSeparator.Length)
				{<commit_msg>use EndsWith instead of IndexOf in ContinueCollectSeparator<commit_after>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;

namespace EtlGate.Core
{
	public interface IDelimitedDataReader
	{
		IEnumerable<Record> ReadFrom(Stream stream, string fieldSeparator = ",", string recordSeparator = "\r\n", bool supportQuotedFields = true, bool hasHeaderRow = false);
	}

	public class DelimitedDataReader : IDelimitedDataReader
	{
		private readonly IStreamTokenizer _streamTokenizer;

		public DelimitedDataReader(IStreamTokenizer streamTokenizer)
		{
			_streamTokenizer = streamTokenizer;
		}

		public IEnumerable<Record> ReadFrom(Stream stream, string fieldSeparator = ",", string recordSeparator = "\r\n", bool supportQuotedFields = true, bool hasHeaderRow = false)
		{
			if (stream == null)
			{
				throw new ArgumentException();
			}

			if (fieldSeparator == null)
			{
				fieldSeparator = "";
			}

			if (recordSeparator == null)
			{
				recordSeparator = "";
			}

			if (fieldSeparator == recordSeparator)
			{
				throw new ParseException("field separator and record separator must be different.");
			}

			var bufferedStream = new BufferedStream(stream, 131072);

			var fieldValues = new List<string>();
			var parseContext = new ParseContext
				                   {
					                   HeaderRowFieldIndexes = new Dictionary<string, int>(),
					                   HasHeaderRow = hasHeaderRow,
					                   RecordSeparator = recordSeparator,
					                   FieldSeparator = fieldSeparator,
					                   IndexOfRecordSeparatorInFieldSeparator = fieldSeparator.IndexOf(recordSeparator, StringComparison.Ordinal),
					                   ReadingHeaderRow = hasHeaderRow,
					                   Handle = StartReadField,
					                   RecordNumber = 1,
					                   SupportQuotedFields = supportQuotedFields,
					                   Capture = new StringBuilder()
				                   };
			var specialTokens = new List<string>();
			if (!String.IsNullOrEmpty(fieldSeparator))
			{
				specialTokens.Add(fieldSeparator);
			}
			if (!String.IsNullOrEmpty(recordSeparator))
			{
				specialTokens.Add(recordSeparator);
			}
			if (supportQuotedFields)
			{
				specialTokens.Add("\"");
			}
			var specialChars = (fieldSeparator + recordSeparator + (supportQuotedFields ? "\"" : "")).Distinct().ToArray();
			foreach (var token in _streamTokenizer.Tokenize(bufferedStream, specialChars))
			{
				var yieldIt = parseContext.Handle(token, fieldValues, parseContext);
				if (yieldIt)
				{
					yield return new Record(fieldValues, parseContext.HeaderRowFieldIndexes);
					fieldValues = parseContext.HasHeaderRow
						              ? new List<string>(parseContext.HeaderRowFieldIndexes.Count)
						              : new List<string>(fieldValues.Count);
					parseContext.Capture.Clear();
				}
			}
			if (parseContext.Handle == ReadQuotedField)
			{
				throw new ParseException("Quoted field does not have a close quote on line " + parseContext.RecordNumber + " field " + parseContext.FieldNumber);
			}
			if (parseContext.ReadingQuotedField && parseContext.Handle == ContinueCollectSeparator)
			{
				ThrowUnescapedQuoteException(parseContext);
			}
			if (parseContext.Capture.Length > 0 ||
			    parseContext.FieldNumber > 0 &&
			    (parseContext.Handle == StartReadField || parseContext.Handle == BranchReadEscapedQuoteOrStartCollectingSeparator))
			{
				if (parseContext.Handle == StartReadField)
				{
					parseContext.IncrementFieldNumber();
				}
				AddFieldValue(fieldValues, parseContext);
			}
			if (fieldValues.Count > 0)
			{
				yield return new Record(fieldValues, parseContext.HeaderRowFieldIndexes);
			}
		}

		private static void AddFieldValue(ICollection<string> fieldValues, ParseContext parseContext)
		{
			var capture = parseContext.Capture;
			var value = capture.ToString();
			fieldValues.Add(value);
			capture.Length = 0;
		}

		private static bool BranchReadEscapedQuoteOrStartCollectingSeparator(Token token, List<string> fieldValues, ParseContext parseContext)
		{
			if (token is DataToken)
			{
				ThrowUnescapedQuoteException(parseContext);
			}

			switch (token.Value)
			{
				case "\"":
					parseContext.Capture.Append('"');
					parseContext.Handle = ReadQuotedField;
					break;
				default:
					if (parseContext.FieldSeparator.Length > 0 && token.Value[0] == parseContext.FieldSeparator[0] ||
					    parseContext.RecordSeparator.Length > 0 && token.Value[0] == parseContext.RecordSeparator[0])
					{
						// possible end of field, more fields follow
						return StartCollectingSeparator(token, fieldValues, parseContext);
					}

					ThrowUnescapedQuoteException(parseContext);
					break;
			}
			return false;
		}

		private static void CheckHeaderRowForDuplicateFieldNames(IEnumerable<string> fieldValues)
		{
			var uniqueHeaderLookup = new HashSet<string>();
			foreach (var field in fieldValues.Where(field => !uniqueHeaderLookup.Add(field)))
			{
				throw new ParseException(String.Format("Header row must not have more than one field with the same name. '{0}' appears more than once in the header row.", field));
			}
		}

		private static bool ContinueCollectSeparator(Token token, List<string> fieldValues, ParseContext parseContext)
		{
			var capture = parseContext.Capture;
			capture.Append(token.Value);
			var potentialSeparator = capture.ToString(capture.Length - parseContext.SeparatorLength, parseContext.SeparatorLength);
			var fieldSeparatorIndex = parseContext.FieldSeparator.Length > 0 && 
				potentialSeparator.EndsWith(parseContext.FieldSeparator, StringComparison.Ordinal) 
				? potentialSeparator.Length - parseContext.FieldSeparator.Length : -1;
			var recordSeparatorIndex = parseContext.RecordSeparator.Length > 0 && 
				potentialSeparator.EndsWith(parseContext.RecordSeparator, StringComparison.Ordinal) 
				? potentialSeparator.Length - parseContext.RecordSeparator.Length : -1;

			if (fieldSeparatorIndex != -1 && recordSeparatorIndex != -1)
			{
				return ReadFirstSeparator(fieldValues, parseContext, fieldSeparatorIndex, recordSeparatorIndex, potentialSeparator);
			}

			if (fieldSeparatorIndex != -1)
			{
				return TryReadFieldSeparator(fieldValues, parseContext, fieldSeparatorIndex, potentialSeparator);
			}

			if (recordSeparatorIndex != -1)
			{
				if (parseContext.IndexOfRecordSeparatorInFieldSeparator <= 0)
				{
					return TryReadRecordSeparator(fieldValues, parseContext, recordSeparatorIndex, potentialSeparator);
				}

				// implicit:	parseContext.IndexOfRecordSeparatorInFieldSeparator > 0
				var potentialFieldSeparator = potentialSeparator.Substring(recordSeparatorIndex - parseContext.IndexOfRecordSeparatorInFieldSeparator);
				if (potentialFieldSeparator.Length >= parseContext.FieldSeparator.Length)
				{
{'fraction_non_alphanumeric': 0.09082252984692009, 'fraction_numerical': 0.004101662638248004, 'mean_word_length': 3.4002578150177247, 'pattern_counts': {'":': 2, '<': 29, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6187976', 'n_tokens_mistral': 4450, 'n_tokens_neox': 3652, 'n_words': 780}