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

namespace LogFlow.Builtins.Inputs
{
	public class FileInput : LogInput, IStartable
	{
		private static readonly Logger Log = LogManager.GetCurrentClassLogger();

		private readonly ConcurrentQueue<string> _changedFiles = new ConcurrentQueue<string>();
		private LineInProcess _lineInProcess;
		private readonly FileSystemWatcher _watcher;

		private readonly string _path;
		private readonly Encoding _encoding;

		public FileInput(string path) : this(path, Encoding.UTF8) { }

		public FileInput(string path, Encoding encoding)
		{
			_path = path;
			_encoding = encoding;

			_watcher = new FileSystemWatcher(GetPath(), GetSearchPattern());
			_watcher.Changed += (sender, args) => AddToQueueWithDuplicationCheck(args.FullPath);
			_watcher.Created += (sender, args) => AddToQueueWithDuplicationCheck(args.FullPath);
		}

		private void AddToQueueWithDuplicationCheck(string fullPath)
		{
			if (!_changedFiles.Contains(fullPath))
			{
				Log.Trace(string.Format("{0}: Enqueuing file '{1}' for processing.", LogContext.LogType, fullPath));
				_changedFiles.Enqueue(fullPath);
			}
		}

		private string GetPath()
		{
			return Path.GetDirectoryName(_path);
		}

		private string GetSearchPattern()
		{
			var fileName = Path.GetFileName(_path);
			return string.IsNullOrWhiteSpace(fileName) ? "*" : fileName.Trim();
		}

		public void Start()
		{
			Log.Info(string.Format("{0}: Starting FileInput.", LogContext.LogType));

			AddCurrentFilesToQueue();
			StartFileSystemWatcher();
		}

		private static string GetPositionKey(string filePath)
		{
			return "position_" + filePath;
		}

		private void StartFileSystemWatcher()
		{
			_watcher.EnableRaisingEvents = true;
			Log.Info(string.Format("{0}: Started FileSystemWatcher for path {1}", LogContext.LogType, _path));
		}

		private void StopFileSystemWatcher()
		{
			_watcher.EnableRaisingEvents = false;
			Log.Info(string.Format("{0}: Stopped FileSystemWatcher for path {1}", LogContext.LogType, _path));
		}

		private void AddCurrentFilesToQueue()
		{
			Log.Info(string.Format("{0}: Adding all current files as changed.", LogContext.LogType));

			foreach (var file in GetCurrentFiles())
			{
				AddToQueueWithDuplicationCheck(file);
			}
		}

		private IEnumerable<string> GetCurrentFiles()
		{
			return Directory.GetFiles(GetPath(), GetSearchPattern()).OrderBy(f => new FileInfo(f).LastWriteTime);
		}

		public void Stop()<commit_msg>Added parameter for recursion in directories<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using NLog;

namespace LogFlow.Builtins.Inputs
{
	public class FileInput : LogInput, IStartable
	{
		private static readonly Logger Log = LogManager.GetCurrentClassLogger();

		private readonly ConcurrentQueue<string> _changedFiles = new ConcurrentQueue<string>();
		private LineInProcess _lineInProcess;
		private readonly FileSystemWatcher _watcher;

		private readonly string _path;
		private readonly Encoding _encoding;

		public FileInput(string path) : this(path, Encoding.UTF8, false) { }

		public FileInput(string path, Encoding encoding, bool includeSubDirectories)
		{
			_path = path;
			_encoding = encoding;

			_watcher = new FileSystemWatcher(GetPath(), GetSearchPattern()) { IncludeSubdirectories = includeSubDirectories };
			_watcher.Changed += (sender, args) => AddToQueueWithDuplicationCheck(args.FullPath);
			_watcher.Created += (sender, args) => AddToQueueWithDuplicationCheck(args.FullPath);
		}

		private void AddToQueueWithDuplicationCheck(string fullPath)
		{
			if (!_changedFiles.Contains(fullPath))
			{
				Log.Trace(string.Format("{0}: Enqueuing file '{1}' for processing.", LogContext.LogType, fullPath));
				_changedFiles.Enqueue(fullPath);
			}
		}

		private string GetPath()
		{
			return Path.GetDirectoryName(_path);
		}

		private string GetSearchPattern()
		{
			var fileName = Path.GetFileName(_path);
			return string.IsNullOrWhiteSpace(fileName) ? "*" : fileName.Trim();
		}

		public void Start()
		{
			Log.Info(string.Format("{0}: Starting FileInput.", LogContext.LogType));

			AddCurrentFilesToQueue();
			StartFileSystemWatcher();
		}

		private static string GetPositionKey(string filePath)
		{
			return "position_" + filePath;
		}

		private void StartFileSystemWatcher()
		{
			_watcher.EnableRaisingEvents = true;
			Log.Info(string.Format("{0}: Started FileSystemWatcher for path {1}", LogContext.LogType, _path));
		}

		private void StopFileSystemWatcher()
		{
			_watcher.EnableRaisingEvents = false;
			Log.Info(string.Format("{0}: Stopped FileSystemWatcher for path {1}", LogContext.LogType, _path));
		}

		private void AddCurrentFilesToQueue()
		{
			Log.Info(string.Format("{0}: Adding all current files as changed.", LogContext.LogType));

			foreach (var file in GetCurrentFiles())
			{
				AddToQueueWithDuplicationCheck(file);
			}
		}

		private IEnumerable<string> GetCurrentFiles()
		{
			var searchOption = _watcher.IncludeSubdirectories ? SearchOption.AllDirectories : SearchOption.TopDirectoryOnly;
			return Directory.GetFiles(GetPath(), GetSearchPattern(), searchOption).OrderBy(f => new FileInfo(f).LastWriteTime);
		}

		public void Stop()
{'fraction_non_alphanumeric': 0.11006982726938626, 'fraction_numerical': 0.0033076074972436605, 'mean_word_length': 4.897074756229686, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, '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': '5775264', 'n_tokens_mistral': 1848, 'n_tokens_neox': 1620, 'n_words': 367}