<commit_before>import pyexcel_io
import pyexcel_xlsx
import os
import csv
import datetime
from etk.document import Document
from typing import List
from io import StringIO
from etk.etk_exceptions import InvalidArgumentsError, InvalidFilePathError


class CsvProcessor(object):
    """
                        heading_row: int,
                        content_start_row: int,
                        heading_columns: (int, int),
                        content_end_row: int,
                        blank_row_ends_content: bool,
                        remove_leading_trailing_whitespace: bool,
                        required_columns: list[str]
                        column_name_prefix: str
    """

    def __init__(self, etk, **mapping_spec) -> None:
        self.etk = etk

        self.heading_row = mapping_spec.get("heading_row")
        if self.heading_row is not None:
            self.heading_row = self.heading_row - 1

        if self.heading_row is not None:
            self.content_start_row = mapping_spec.get("content_start_row", self.heading_row + 2) - 1

        if self.heading_row is None:
            self.content_start_row = mapping_spec.get("content_start_row", 1) - 1

        # how about if heading_col is not present? read until first empty cell?
        self.heading_columns = mapping_spec.get("heading_columns")

        if self.heading_columns is not None:
            self.heading_columns = (self.heading_columns[0] - 1, self.heading_columns[1])

        # if not present, default read until an empty row
        self.content_end_row = mapping_spec.get("content_end_row")

        # if set to false, read until EOF
        self.blank_row_ends_content = mapping_spec.get("ends_with_blank_row", True)

        # remove all white space of the value
        self.remove_leading_trailing_whitespace = \
            mapping_spec.get("remove_leading_trailing_whitespace", True)

        # remove all empty rows before the content
        self.remove_leading_empty_rows = \
            mapping_spec.get("remove_leading_empty_rows", True)

        self.required_columns = mapping_spec.get("required_columns")

        if mapping_spec.get("column_name_prefix"):
            self.column_name_prefix = mapping_spec.get("column_name_prefix")
        else:
            self.column_name_prefix = "C"

        self._get_data_function = {
            ".tab": pyexcel_io.get_data,
            ".csv": pyexcel_io.get_data,
            ".tsv": pyexcel_io.get_data,
            ".xls": pyexcel_xlsx.get_data,
            ".xlsx": pyexcel_xlsx.get_data
        }

    def tabular_extractor(self, table_str: str = None, filename: str = None,
                          file_content=None,
                          file_type=None,
                          sheet_name: str = None,
                          dataset: str = None,
                          nested_key: str = None,
                          doc_id_field: str = None) -> List[Document]:
        data = list()

        if table_str is not None and filename is not None:
            raise InvalidArgumentsError(message="for arguments 'table_str' and 'filename', please specify only one "
                                                "argument!")

        elif table_str is not None:
            f = StringIO(table_str)
            reader = csv.reader(f, delimiter=',')
            for row in reader:
                data.append(row)
        elif filename is not None:
            # always read the entire file first
            fn, extension = os.path.splitext(filename)
            extension = extension.lower()

            if extension in self._get_data_function:
                get_data = self._get_data_function[extension]
            else:
                raise InvalidFilePathError("file extension can not read")

            try:
                if file_content and file_type:
                    data = get_data(file_content, file_type=file_type, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="utf-8")
                else:
                    data = get_data(filename, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="utf-8")
            except:
                try:
                    data = get_data(filename, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="latin_1")
                except:
                    data = get_data(filename, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="utf-8-sig")

            if extension == '.xls' or extension == '.xlsx':
                if sheet_name is None:
                    sheet_name = list(data.keys())[0]

                data = data[sheet_name]
            else:
                if extension == '.tab':
                    data = data['tsv']
                else:
                    file_name = fn.split('/')[-1] + extension
                    data = data[file_name]

        table_content, heading = self.content_recognizer(data)

        return self.create_documents(rows=table_content,
                                     heading=heading,
                                     file_name=filename,
                                     dataset=dataset,
                                     nested_key=nested_key,
                                     doc_id_field=doc_id_field)

    def content_recognizer(self, data: List[List[str]]) -> tuple((List[List[str]], List[str])):
        heading = list()
        # process heading
        if self.heading_row is not None:
            heading, col_start, col_end = self.process_heading(data[self.heading_row])
            # if heading_row is specified, discards/overwrite the heading_columns
            self.heading_columns = (col_start, col_end)
        else:
            heading = None

        # handle row first:
        if self.content_end_row is None:
            data = data[self.content_start_row:]

        if self.content_end_row is not None:
            data = data[self.content_start_row:self.content_end_row]<commit_msg>remove dependency on file extensions in csv processor<commit_after>import pyexcel_io
import pyexcel_xlsx
import os
import csv
import datetime
from etk.document import Document
from typing import List
from io import StringIO
from etk.etk_exceptions import InvalidArgumentsError, InvalidFilePathError


class CsvProcessor(object):
    """
                        heading_row: int,
                        content_start_row: int,
                        heading_columns: (int, int),
                        content_end_row: int,
                        blank_row_ends_content: bool,
                        remove_leading_trailing_whitespace: bool,
                        required_columns: list[str]
                        column_name_prefix: str
    """

    def __init__(self, etk, **mapping_spec) -> None:
        self.etk = etk

        self.heading_row = mapping_spec.get("heading_row")
        if self.heading_row is not None:
            self.heading_row = self.heading_row - 1

        if self.heading_row is not None:
            self.content_start_row = mapping_spec.get("content_start_row", self.heading_row + 2) - 1

        if self.heading_row is None:
            self.content_start_row = mapping_spec.get("content_start_row", 1) - 1

        # how about if heading_col is not present? read until first empty cell?
        self.heading_columns = mapping_spec.get("heading_columns")

        if self.heading_columns is not None:
            self.heading_columns = (self.heading_columns[0] - 1, self.heading_columns[1])

        # if not present, default read until an empty row
        self.content_end_row = mapping_spec.get("content_end_row")

        # if set to false, read until EOF
        self.blank_row_ends_content = mapping_spec.get("ends_with_blank_row", True)

        # remove all white space of the value
        self.remove_leading_trailing_whitespace = \
            mapping_spec.get("remove_leading_trailing_whitespace", True)

        # remove all empty rows before the content
        self.remove_leading_empty_rows = \
            mapping_spec.get("remove_leading_empty_rows", True)

        self.required_columns = mapping_spec.get("required_columns")

        if mapping_spec.get("column_name_prefix"):
            self.column_name_prefix = mapping_spec.get("column_name_prefix")
        else:
            self.column_name_prefix = "C"

        self._get_data_function = {
            ".csv": pyexcel_io.get_data,
            ".tsv": pyexcel_io.get_data,
            ".xls": pyexcel_xlsx.get_data,
            ".xlsx": pyexcel_xlsx.get_data
        }

    def tabular_extractor(self, table_str: str = None, filename: str = None,
                          file_content=None,
                          file_type=None,
                          sheet_name: str = None,
                          dataset: str = None,
                          nested_key: str = None,
                          doc_id_field: str = None) -> List[Document]:
        """
        Read the input file/content and return a list of Document(s)
        Args:
            table_str: use this parameter, if you are 100% sure that the content is a csv
            filename: use this parameter if the file extension is one of tab, csv, tsv, xls, xlsx
            file_content: if the input has some arbitrary extension, read it yourself and pass the contents along
            file_type: use this parameter with file_content, can be tsv, csv, etc
            sheet_name: sheet name as in xls or xlsx files
            dataset: user provided string to be added to output Document(s)
            nested_key: user provided string to be added to output Document(s)
            doc_id_field: specify this field(should be present in the input file), its value will be used as doc_id

        Returns: List[Document]

        """
        data = list()

        if table_str is not None and filename is not None:
            raise InvalidArgumentsError(message="for arguments 'table_str' and 'filename', please specify only one "
                                                "argument!")

        elif table_str is not None:
            f = StringIO(table_str)
            reader = csv.reader(f, delimiter=',')
            for row in reader:
                data.append(row)
        elif filename is not None:
            # always read the entire file first
            fn, extension = os.path.splitext(filename)
            extension = extension.lower()

            if extension in self._get_data_function:
                get_data = self._get_data_function[extension]
            else:
                # in pyexcel we trust
                # if there is an extension we have not mapped, just let pyexcel figure it out
                get_data = pyexcel_io.get_data

            try:
                if file_content and file_type:
                    data = get_data(file_content, file_type=file_type, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="utf-8")
                else:
                    data = get_data(filename, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="utf-8")
            except:
                try:
                    data = get_data(filename, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="latin_1")
                except:
                    data = get_data(filename, auto_detect_datetime=False,
                                    auto_detect_float=False, encoding="utf-8-sig")

            if extension == '.xls' or extension == '.xlsx':
                if sheet_name is None:
                    sheet_name = list(data.keys())[0]

                data = data[sheet_name]
            else:
                data = data[file_type] if file_type else data[fn.split('/')[-1] + extension]

        table_content, heading = self.content_recognizer(data)

        return self.create_documents(rows=table_content,
                                     heading=heading,
                                     file_name=filename,
                                     dataset=dataset,
                                     nested_key=nested_key,
                                     doc_id_field=doc_id_field)

    def content_recognizer(self, data: List[List[str]]) -> tuple((List[List[str]], List[str])):
        heading = list()
        # process heading
        if self.heading_row is not None:
            heading, col_start, col_end = self.process_heading(data[self.heading_row])
            # if heading_row is specified, discards/overwrite the heading_columns
            self.heading_columns = (col_start, col_end)
        else:
            heading = None

        # handle row first:
        if self.content_end_row is None:
            data = data[self.content_start_row:]

        if self.content_end_row is not None:
            data = data[self.content_start_row:self.content_end_row]
{'fraction_non_alphanumeric': 0.07477064220183487, 'fraction_numerical': 0.0023700305810397553, 'mean_word_length': 1.729188399749635, 'pattern_counts': {'":': 9, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6233006', 'n_tokens_mistral': 3514, 'n_tokens_neox': 3366, 'n_words': 870}