<commit_before>package dsc

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"database/sql"
	"fmt"
	"github.com/viant/toolbox"
	"github.com/viant/toolbox/storage"
	"io"
	"io/ioutil"
	"net/url"
	"os"
	"path"
	"reflect"
	"strings"
)

var defaultPermission os.FileMode = 0644

//FileManager represents a line delimiter, JSON file manager.
// Current implementation is brute/force full file scan on each operation.
// Do not use it on production, it is not performance optimized.
// Idea behind is to be able to load data log and test it.
// You can easily add other managers providing your custom encoder and decoder factories i.e. protobuf, avro.
type FileManager struct {
	*AbstractManager
	service             storage.Service
	useGzipCompressions bool
	hasHeaderLine       bool
	delimiter           string
	encoderFactory      toolbox.EncoderFactory
	decoderFactory      toolbox.DecoderFactory
}

func (m *FileManager) Init() error {
	var baseUrl = m.Config().Get("url")
	parsedUrl, err := url.Parse(baseUrl)
	if err != nil {
		return err
	}
	extension := m.Config().Get("ext")
	m.useGzipCompressions = extension == "gzip"
	switch parsedUrl.Scheme {
	case "file":

	case "gs":
		fallthrough
	case "s3":

		credential := m.Config().Get("credential")
		service, err := storage.NewServiceForURL(baseUrl, credential)
		if err != nil {
			return err
		}
		m.service.Register(parsedUrl.Scheme, service)

	default:
		return fmt.Errorf("unsupported scheme: %v", parsedUrl.Scheme)
	}
	return nil

}

func (m *FileManager) convertIfNeeded(source interface{}) interface{} {
	if toolbox.IsTime(source) {
		dateLayout := m.config.GetDateLayout()
		return toolbox.AsTime(source, dateLayout).Format(dateLayout)
	}
	sourceValue := reflect.ValueOf(toolbox.DereferenceValue(source))
	switch sourceValue.Kind() {

	case reflect.Slice:
		if sourceValue.Len() == 0 {
			return nil
		}

	case reflect.Map:
		if sourceValue.Len() == 0 {
			return nil
		}

	case reflect.Struct:
		var values = make(map[string]interface{})
		fieldsMappingByField := toolbox.NewFieldSettingByKey(source, "fieldName")

		toolbox.ProcessStruct(source,
			func(filed reflect.StructField, value interface{}) {
				mapping := fieldsMappingByField[filed.Name]
				column, found := mapping["column"]
				if !found {
					column = filed.Name
				}
				values[column] = m.convertIfNeeded(value)
			})
		if len(values) == 0 {
			return toolbox.AsString(source)
		}
		return values
	}
	return source
}

func getTableURL(manager Manager, table string) string {
	tableFile := table + "." + path.Join(manager.Config().Get("ext"))
	return path.Join(manager.Config().Get("url"), tableFile)
}

func (m *FileManager) encodeRecord(record map[string]interface{}, table string) (string, error) {
	var buffer = new(bytes.Buffer)
	var encoder = m.encoderFactory.Create(buffer)

	if m.delimiter != "" {
		descriptor := m.TableDescriptorRegistry().Get(table)
		encoder.Encode(descriptor.Columns)
	}

	err := encoder.Encode(&record)
	if err != nil {
		return "", fmt.Errorf("failed to encode record: %v due to ", err)
	}
	result := string(buffer.Bytes())
	result = strings.Replace(result, "\n", "", len(result)) + "\n"
	return result, nil
}<commit_msg> updated  process map<commit_after>package dsc

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"database/sql"
	"fmt"
	"github.com/viant/toolbox"
	"github.com/viant/toolbox/storage"
	"io"
	"io/ioutil"
	"net/url"
	"os"
	"path"
	"reflect"
	"strings"
)

var defaultPermission os.FileMode = 0644

//FileManager represents a line delimiter, JSON file manager.
// Current implementation is brute/force full file scan on each operation.
// Do not use it on production, it is not performance optimized.
// Idea behind is to be able to load data log and test it.
// You can easily add other managers providing your custom encoder and decoder factories i.e. protobuf, avro.
type FileManager struct {
	*AbstractManager
	service             storage.Service
	useGzipCompressions bool
	hasHeaderLine       bool
	delimiter           string
	encoderFactory      toolbox.EncoderFactory
	decoderFactory      toolbox.DecoderFactory
}

func (m *FileManager) Init() error {
	var baseUrl = m.Config().Get("url")
	parsedUrl, err := url.Parse(baseUrl)
	if err != nil {
		return err
	}
	extension := m.Config().Get("ext")
	m.useGzipCompressions = extension == "gzip"
	switch parsedUrl.Scheme {
	case "file":

	case "gs":
		fallthrough
	case "s3":

		credential := m.Config().Get("credential")
		service, err := storage.NewServiceForURL(baseUrl, credential)
		if err != nil {
			return err
		}
		m.service.Register(parsedUrl.Scheme, service)

	default:
		return fmt.Errorf("unsupported scheme: %v", parsedUrl.Scheme)
	}
	return nil

}

func (m *FileManager) convertIfNeeded(source interface{}) interface{} {
	if toolbox.IsTime(source) {
		dateLayout := m.config.GetDateLayout()
		return toolbox.AsTime(source, dateLayout).Format(dateLayout)
	}
	sourceValue := reflect.ValueOf(toolbox.DereferenceValue(source))
	switch sourceValue.Kind() {

	case reflect.Slice:
		if sourceValue.Len() == 0 {
			return nil
		}

	case reflect.Map:
		if sourceValue.Len() == 0 {
			return nil
		}

	case reflect.Struct:
		var values = make(map[string]interface{})
		fieldsMappingByField := toolbox.NewFieldSettingByKey(source, "fieldName")

		toolbox.ProcessStruct(source,
			func(filed reflect.StructField, value reflect.Value) error {
				mapping := fieldsMappingByField[filed.Name]
				column, found := mapping["column"]
				if !found {
					column = filed.Name
				}
				values[column] = m.convertIfNeeded(value.Interface())
				return nil
			})
		if len(values) == 0 {
			return toolbox.AsString(source)
		}
		return values
	}
	return source
}

func getTableURL(manager Manager, table string) string {
	tableFile := table + "." + path.Join(manager.Config().Get("ext"))
	return path.Join(manager.Config().Get("url"), tableFile)
}

func (m *FileManager) encodeRecord(record map[string]interface{}, table string) (string, error) {
	var buffer = new(bytes.Buffer)
	var encoder = m.encoderFactory.Create(buffer)

	if m.delimiter != "" {
		descriptor := m.TableDescriptorRegistry().Get(table)
		encoder.Encode(descriptor.Columns)
	}

	err := encoder.Encode(&record)
	if err != nil {
		return "", fmt.Errorf("failed to encode record: %v due to ", err)
	}
	result := string(buffer.Bytes())
	result = strings.Replace(result, "\n", "", len(result)) + "\n"
	return result, nil
}
{'fraction_non_alphanumeric': 0.12837732312978292, 'fraction_numerical': 0.002498828674059035, 'mean_word_length': 4.018808777429467, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, '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': '4390692', 'n_tokens_mistral': 2218, 'n_tokens_neox': 2051, 'n_words': 538}