<commit_before>"""
Filters for the Humanitarian Exchange Language (HXL) v1.0
<NAME>
Started October 2014

License: Public Domain
Documentation: https://github.com/HXLStandard/libhxl-python/wiki
"""

import sys, re, six, abc, copy
import dateutil.parser

import hxl


#
# Filter-specific exception
#
class HXLFilterException(hxl.common.HXLException):
    pass

#
# Base class for filters

class AbstractFilter(hxl.model.Dataset):
    """
    Abstract base class for composable filters.

    This class stores the upstream source, and provides a
    filter_columns() method that child classes can implement. It will
    be called precisely once for each instantiation, giving the child
    a chance to provide a different set of columns than those in the
    source.
    """

    __metaclass__ = abc.ABCMeta

    def __init__(self, source):
        """
        Construct a new abstract filter.
        @param source the source dataset
        """
        self.source = source
        self.filtered_columns = None

    @property
    def columns(self):
        if self.filtered_columns is None:
            self.filtered_columns = self.filter_columns()
        return self.filtered_columns

    def filter_columns(self):
        """
        Return a new list of columns for the filtered dataset.
        @return a list of hxl.model.Column objects
        """
        return self.source.columns

    
class AbstractStreamingFilter(AbstractFilter):
    """
    Abstract base class for streaming filters.

    A streaming filter processes one row at a time.  It can skip rows,
    but it never reorders them.  Child classes will implement the
    filter_columns() method from the AbstractFilter class, as well as
    the filter_row(row) method from this class.
    """

    __metaclass__ = abc.ABCMeta

    def __init__(self, source):
        super(AbstractStreamingFilter, self).__init__(source)

    @abc.abstractmethod
    def filter_row(self, row):
        """
        Filter a single row of data.
        A return value of None will cause the row to be skipped.

        @param row A hxl.model.Row object to filter.
        @return An array of new values (not a Row object) or None to skip the row.
        """
        return row.values

    def __iter__(self):
        return AbstractStreamingFilter.Iterator(self)

    class Iterator:
        """
        Iterator to return the filtered rows.
        """

        def __init__(self, outer):
            self.outer = outer
            self.source_iter = iter(self.outer.source)
            self.row_number = -1

        def __iter__(self):
            return self

        def __next__(self):
            # call this here, in case it caches any useful information
            columns = self.outer.columns
            for row in self.source_iter:
                # get a new list of filtered values
                values = self.outer.filter_row(row)
                if values is not None:
                    # keep looping if filter_row(row) returned None
                    self.row_number += 1
                    # create a new Row object
                    return hxl.model.Row(columns, values, self.row_number)
            # if we've finished the iteration, then we're out of rows, so stop
            raise StopIteration()

        next = __next__


class AbstractCachingFilter(AbstractFilter):
    """
    Abstract base class for caching filters.

    A caching filter reads all of the input data first,
    then returns it, possibly in a different order
    or even completely transformed.
    """

    __metaclass__ = abc.ABCMeta

    def __init__(self, source):
        super(AbstractCachingFilter, self).__init__(source)

    def filter_rows(self):
        return self.source.values

    def __iter__(self):
        return AbstractCachingFilter.Iterator(self)

    class Iterator:

        def __init__(self, outer):
            self.outer = outer
            self.values_iter = None
            self.row_number = -1

        def __iter__(self):
            return self

        def __next__(self):
            if self.values_iter is None:
                self.values_iter = iter(outer.filter_rows())
            self.row_number += 1
            return hxl.model.Row(outer.columns, next(self.values_iter), self.row_number)
        
        next = __next__


#
# Filter classes
#

class AddColumnsFilter(AbstractStreamingFilter):
    """
    Composable filter class to add constant values to every row of a HXL dataset.

    This is the class supporting the hxladd command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.
    """

    def __init__(self, source, specs, before=False):
        """
        @param source a HXL data source
        @param specs a sequence of pairs of Column objects and constant values
        @param before True to add new columns before existing ones
        """
        super(AddColumnsFilter, self).__init__(source)
        if isinstance(specs, six.string_types):
            specs = [specs]
        self.specs = [AddColumnsFilter.parse_spec(spec) for spec in specs]
        self.before = before
        self.const_values = None

    def filter_columns(self):
        new_columns = [spec[0] for spec in self.specs]
        if self.before:
            new_columns = new_columns + self.source.columns
        else:
            new_columns = self.source.columns + new_columns
        self.const_values = [spec[1] for spec in self.specs]
        return new_columns

    def filter_row(self, row):
        values = copy.copy(row.values)
        if self.before:
            return self.const_values + values
        else:
            return values + self.const_values

    VALUE_PATTERN = r'^\s*(?:([^#]*)#)?({token}(?:\s*\+{token})*)=(.*)\s*$'.format(token=hxl.common.TOKEN)

    @staticmethod
    def parse_spec(spec):
        if not isinstance(spec, six.string_types):
            return spec
        result = re.match(AddColumnsFilter.VALUE_PATTERN, spec)
        if result:
            header = result.group(1)
            tag = '#' + result.group(2)
            value = result.group(3)
            return (hxl.model.Column(tag=tag, header=header), value)
        else:
            raise HXLFilterException("Badly formatted new-column spec: " + spec)


class AppendFilter(hxl.model.Dataset):
    """Composable filter class to concatenate two datasets."""

    def __init__(self, source, append_source, add_columns=True):
        """
        Constructor
        @param source the HXL data source
        @param append_source the HXL source to append
        @param add_columns flag for adding extra columns in append_source but not source (default True)
        """
        self.source = source
        self.append_source = append_source
        self.add_columns = add_columns
        self._saved_columns = None
        self._column_positions = None
        self._template_row = None

    @property
    def columns(self):
        if self._saved_columns is None:
            # Merge the columns from the second source into the first,
            # appending if necessary (display tag is the key)
            saved_columns = list(self.source.columns)
            column_positions = {}
            original_tags = self.source.display_tags

            # see if there's a corresponding column in the source
            for i, column in enumerate(self.append_source.columns):
                for j, tag in enumerate(original_tags):
                    if tag and (column.display_tag == tag):
                        # yes, there is one; clear it, so it's not reused
                        original_tags[j] = None
                        column_positions[i] = j
                        break
                else:
                    # no -- we need to add a new column
                    if self.add_columns:
                        column_positions[i] = len(saved_columns)
                        saved_columns.append(copy.deepcopy(column))
                    else:
                        column_positions[i] = None
            self._column_positions = column_positions
            self._saved_columns = saved_columns

            # make an empty template for each row
            self._template_row = [''] * len(saved_columns)

        # return the (usually cached) columns
        return self._saved_columns

    def __iter__(self):
        self.columns # make sure this is triggered first
        return AppendFilter.Iterator(self)

    class Iterator:
        """Custom iterator to return the contents of both sources, in sequence."""

        def __init__(self, outer):
            self.outer = outer
            self.source_iter = iter(outer.source)
            self.append_iter = iter(outer.append_source)

        def __iter__(self):
            return self

        def __next__(self):

            # Read from the original source first
            if self.source_iter is not None:
                try:
                    row_in = next(self.source_iter)
                    row_out = copy.deepcopy(row_in)
                    row_out.values = copy.deepcopy(self.outer._template_row)
                    for i, value in enumerate(row_in):
                        row_out.values[i] = row_in.values[i]
                    return row_out
                except StopIteration:
                    # don't let the end of the first source finish the iteration
                    self.source_iter = None

            # Fall through to the append source
            row_in = next(self.append_iter)
            row_out = copy.deepcopy(row_in)
            row_out.values = copy.deepcopy(self.outer._template_row)
            for i, value in enumerate(row_in):
                pos = self.outer._column_positions[i]
                if pos is not None:
                    row_out.values[pos] = value
            return row_out

        next = __next__

        
class CacheFilter(hxl.model.Dataset):
    """Composable filter class to cache HXL data in memory."""

    def __init__(self, source, max_rows=None):
        """
        Constructor
        @param max_rows If >0, maximum number of rows to cache.
        """
        self.source = source
        self.max_rows = max_rows
        self.cached_columns = copy.deepcopy(source.columns)
        self.cached_rows = [copy.deepcopy(row) for row in source]
        self.overflow = False

    @property
    def columns(self):
        return self.cached_columns

    def __iter__(self):
        return iter(self.cached_rows)

    def _load(self):
        if self.cached_rows is None:
            self.cached_rows = []
            self.cached_columns = copy.deepcopy(self.source.columns)
            row_count = 0
            for row in self.source:
                row_count += 1
                if self.max_rows > 1 and row_count >= self.max_rows:
                    self.overflow = True
                    break
                else:
                    self.cached_rows.append(copy.deepcopy(row))


class CleanDataFilter(AbstractStreamingFilter):
    """
    Filter for cleaning values in HXL data.
    Can normalise whitespace, convert to upper/lowercase, and fix dates and numbers.
    TODO: clean up lat/lon coordinates
    """

    def __init__(self, source, whitespace=False, upper=[], lower=[], date=[], number=[]):
        """
        Construct a new data-cleaning filter.
        @param source the HXLDataSource
        @param whitespace list of TagPatterns for normalising whitespace, or True to normalise all.
        @param upper list of TagPatterns for converting to uppercase, or True to convert all.
        @param lower list of TagPatterns for converting to lowercase, or True to convert all.
        @param lower list of TagPatterns for normalising dates, or True to normalise all ending in "_date"
        @param lower list of TagPatterns for normalising numbers, or True to normalise all ending in "_num"
        """
        super(CleanDataFilter, self).__init__(source)
        self.whitespace = whitespace
        self.upper = upper
        self.lower = lower
        self.date = date
        self.number = number

    def filter_row(self, row):
        """Clean up values and pass on the row data."""
        columns = self.columns
        values = copy.copy(row.values)
        for i in range(min(len(values), len(columns))):
            values[i] = self._clean_value(values[i], columns[i])
        return values

    def _clean_value(self, value, column):
        """Clean a single HXL value."""

        value = str(value)

        # Whitespace (-w or -W)
        if self._match_patterns(self.whitespace, column):
            value = re.sub('^\s+', '', value)
            value = re.sub('\s+$', '', value)
            value = re.sub('\s+', ' ', value)

        # Uppercase (-u)
        if self._match_patterns(self.upper, column):
            if sys.version_info[0] > 2:
                value = value.upper()
            else:
                value = value.decode('utf8').upper().encode('utf8')

        # Lowercase (-l)
        if self._match_patterns(self.lower, column):
            if sys.version_info[0] > 2:
                value = value.lower()
            else:
                value = value.decode('utf8').lower().encode('utf8')

        # Date (-d or -D)
        if self._match_patterns(self.date, column, '_date'):
            if value:
                value = dateutil.parser.parse(value).strftime('%Y-%m-%d')

        # Number (-n or -N)
        if self._match_patterns(self.number, column, '_num') and re.match('\d', value):
            if value:
                value = re.sub('[^\d.]', '', value)
                value = re.sub('^0+', '', value)
                value = re.sub('(\..*)0+$', '\g<1>', value)
                value = re.sub('\.$', '', value)

        return value

    def _match_patterns(self, patterns, column, extension=None):
        """Test if a column matches a list of patterns."""
        if not patterns:
            return False
        elif patterns is True:
            # if there's an extension specific like "_date", must match it
            return (column.tag and (not extension or column.tag.endswith(extension)))
        else:
            for pattern in patterns:
                if pattern.match(column):
                    return True
            return False


class ColumnFilter(AbstractStreamingFilter):
    """
    Composable filter class to filter columns in a HXL dataset.

    Usage:

    <pre>
    # blacklist columns
    hxl.data(url).without_columns('contact+email')

    # whitelist columns
    hxl.data(url).with_columns(['org', 'sector', 'adm1'])
    </pre>
    """

    def __init__(self, source, include_tags=[], exclude_tags=[]):
        """
        @param source a HXL data source
        @param include_tags a whitelist of TagPattern objects to include
        @param exclude_tags a blacklist of TagPattern objects to exclude
        """
        super(ColumnFilter, self).__init__(source)
        self.include_tags = hxl.model.TagPattern.parse_list(include_tags)
        self.exclude_tags = hxl.model.TagPattern.parse_list(exclude_tags)
        self.indices = [] # saved indices for columns to include

    def filter_columns(self):
        """
        Remove any columns in the blacklist or not in the whitelist.
        Save the indices in self.cached_indices for row filtering.
        """
        columns_in = self.source.columns
        columns_out = []
        for i in range(len(columns_in)):
            if self._test_column(columns_in[i]):
                columns_out.append(copy.deepcopy(columns_in[i]))
                self.indices.append(i) # save index to avoid retesting for data
        return columns_out

    def filter_row(self, row):
        """Remove values from a row for any column that's been removed."""
        values = []
        for i in self.indices:
            values.append(row.values[i])
        return values

    def _test_column(self, column):
        """
        Test whether a column should be included in the output.
        If there is a whitelist, it must be in the whitelist; if there is a blacklist, it must not be in the blacklist.
        """
        if self.exclude_tags:
            # blacklist
            for pattern in self.exclude_tags:
                if pattern.match(column):
                    # fail as soon as we match an excluded pattern
                    return False

        if self.include_tags:
            # whitelist
            for pattern in self.include_tags:
                if pattern.match(column):
                    # succeed as soon as we match an included pattern
                    return True
            # fail if there was a whitelist and we didn't match
            return False
        else:
            # no whitelist
            return True


class CountFilter(hxl.model.Dataset):
    """
    Composable filter class to aggregate rows in a HXL dataset.

    This is the class supporting the hxlcount command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.

    WARNING: this filter reads the entire source dataset before
    producing output, and may need to hold a large amount of data in
    memory, depending on the number of unique combinations counted.

    Usage:

    <pre>
    filter = CountFilter(source, patterns=['#org', '#sector'])
    </pre>

    or

    <pre>
    filter = source.count(['#org', '#sector'])
    </pre>
    """

    def __init__(self, source, patterns, aggregate_pattern=None):
        """
        Constructor
        @param source the HXL data source
        @param patterns a list of strings or TagPattern objects that form a unique key together
        @param aggregate_pattern an optional tag pattern calculating numeric aggregate values.
        """
        self.source = source
        self.patterns = hxl.model.TagPattern.parse_list(patterns)
        self.aggregate_pattern = hxl.model.TagPattern.parse(aggregate_pattern) if aggregate_pattern else None
        self._saved_columns = None

    @property
    def columns(self):
        """
        @return the column definitions used in the aggregation report
        """
        if self._saved_columns is None:
            cols = []
            for pattern in self.patterns:
                column = pattern.find_column(self.source.columns)
                if column is not None:
                    header = column.header
                else:
                    header = None
                cols.append(hxl.model.Column(tag=pattern.tag, attributes=pattern.include_attributes, header=header))
            cols.append(hxl.model.Column.parse('#meta+count', 'Count'))
            if self.aggregate_pattern is not None:
                cols.append(hxl.model.Column.parse('#meta+sum', header='Sum'))
                cols.append(hxl.model.Column.parse('#meta+average', header='Average (mean)'))
                cols.append(hxl.model.Column.parse('#meta+min', header='Minimum value'))
                cols.append(hxl.model.Column.parse('#meta+max', header='Maximum value'))
            self._saved_columns = cols
        return self._saved_columns

    def __iter__(self):
        return CountFilter.Iterator(self)

    class Iterator:

        def __init__(self, outer):
            self.outer = outer
            self.iterator = iter(outer.source)
            self.aggregate_iter = None

        def __iter__(self):
            return self

        def __next__(self):
            """
            @return the next row of aggregated data.
            """
            if self.aggregate_iter is None:
                self._aggregate()
            # Write the stats, sorted in value order
            aggregate = next(self.aggregate_iter)
            values = list(aggregate[0])
            values.append(aggregate[1].count)
            if self.outer.aggregate_pattern:
                if aggregate[1].seen_numbers:
                    values.append(aggregate[1].sum)
                    values.append(aggregate[1].average)
                    values.append(aggregate[1].min)
                    values.append(aggregate[1].max)
                else:
                    values = values + ([''] * 4)

            row = hxl.model.Row(self.outer.columns)
            row.values = values
            return row

        next = __next__

        def _aggregate(self):
            """
            Read the entire source dataset and produce saved aggregate data.
            """
            aggregators = {}
            for row in self.iterator:
                values = [str(pattern.get_value(row)) for pattern in self.outer.patterns]
                if values:
                    key = tuple(values)
                    if not key in aggregators:
                        aggregators[key] = CountFilter.Aggregator(self.outer.aggregate_pattern)
                    aggregators[key].add(row)
            self.aggregate_iter = iter(sorted(aggregators.items()))

    class Aggregator(object):
        """
        Class to collect aggregates for a single combination.

        Currently calculates count, sum, average, min, and max
        """

        def __init__(self, pattern):
            """
            Constructor
            @param pattern the HXL tag being counted in the row.
            """
            self.pattern = pattern
            self.count = 0
            self.sum = 0.0
            self.average = 0.0
            self.min = None
            self.max = None
            self.seen_numbers = False

        def add(self, row):
            """
            Add a new row of data to the aggregator.
            """
            self.count += 1
            if self.pattern:
                value = self.pattern.get_value(row)
                if value:
                    try:
                        n = float(value)
                        self.sum += n
                        self.average = self.sum / self.count
                        if self.min is None or n < self.min:
                            self.min = n
                        if self.max is None or n > self.max:
                            self.max = n
                        self.seen_numbers = True
                    except:
                        pass


class DeduplicationFilter(AbstractStreamingFilter):
    """
    Composable filter to deduplicate a HXL dataset.

    Supports the hxldedup command-line script.

    TODO: add more-sophisticated matching, edit distance, etc.
    """

    def __init__(self, source, patterns=None):
        """
        Constructor
        @param source the upstream source dataset
        @param patterns if provided, a list of tag patterns for columns to use for uniqueness testing.
        """
        super(DeduplicationFilter, self).__init__(source)
        self.patterns = hxl.model.TagPattern.parse_list(patterns)
        self.seen_map = set() # row signatures that we've seen so far

    def filter_row(self, row):
        """Filter out any rows we've seen before."""
        if not row:
            return None
        key = self._make_key(row)
        if key in self.seen_map:
            return None
        # if we get to here, we haven't seen the row before
        self.seen_map.add(key)
        return copy.copy(row.values)

    def _is_key(self, col):
        """Check if a column is part of the key for deduplication."""
        if self.patterns:
            for pattern in self.patterns:
                if pattern.match(col):
                    return True
            return False
        else:
            return True

    def _make_key(self, row):
        """Create a tuple key for a row."""
        key = []
        for i, value in enumerate(row.values):
            if self._is_key(row.columns[i]):
                key.append(hxl.common.normalise_string(value))
        return tuple(key)

        
class MergeDataFilter(AbstractStreamingFilter):
    """
    Composable filter class to merge values from two HXL datasets.

    This is the class supporting the hxlmerge command-line utility.

    Warning: this filter may store a large amount of data in memory, depending on the merge.

    Usage:

    <pre>
    hxl.data(url).merge(merge_source=merge_source, keys='adm1_id', tags='adm1')
    </pre>
    """

    def __init__(self, source, merge_source, keys, tags, replace=False, overwrite=False):
        """
        Constructor.
        @param source the HXL data source.
        @param merge_source a second HXL data source to merge into the first.
        @param keys the shared key hashtags to use for the merge
        @param tags the tags to include from the second dataset
        """
        super(MergeDataFilter, self).__init__(source)
        self.merge_source = merge_source
        self.keys = hxl.model.TagPattern.parse_list(keys)
        self.merge_tags = hxl.model.TagPattern.parse_list(tags)
        self.replace = replace
        self.overwrite = overwrite

        self.merge_map = None

    def filter_columns(self):
        """Filter the columns to add newly-merged ones."""
        new_columns = []
        for pattern in self.merge_tags:
            if self.replace and pattern.find_column(self.source.columns):
                # will use existing column
                continue
            else:
                column = pattern.find_column(self.merge_source.columns)
                if column:
                    header = column.header
                else:
                    header = None
                new_columns.append(hxl.model.Column(tag=pattern.tag, attributes=pattern.include_attributes, header=header))
        return self.source.columns + new_columns

    def filter_row(self, row):
        """Set up a merged data row, replacing existing values if requested."""
        
        # First, check if we already have the merge map, and read it if not
        if self.merge_map is None:
            self.merge_map = self._read_merge()

        # Make a copy of the values
        values = copy.copy(row.values)

        # Look up the merge values, based on the --keys
        merge_values = self.merge_map.get(self._make_key(row), {})

        # Go through the merge tags
        for pattern in self.merge_tags:
            # Try to substitute in place?
            if self.replace:
                index = pattern.find_column_index(self.source.columns)
                if index is not None:
                    if self.overwrite or not row.values[index]:
                        values[index] = merge_values.get(pattern)
                    continue

            # otherwise, fall through
            values.append(merge_values.get(pattern, ''))
        return values

    def _make_key(self, row):
        """
        Make a tuple key for a row.
        """
        values = []
        for pattern in self.keys:
            values.append(hxl.common.normalise_string(pattern.get_value(row)))
        return tuple(values)

    def _read_merge(self):
        """
        Read the second (merging) dataset into memory.
        Stores only the values necessary for the merge.
        @return a map of merge values
        """
        merge_map = {}
        for row in self.merge_source:
            values = {}
            for pattern in self.merge_tags:
                values[pattern] = pattern.get_value(row)
            merge_map[self._make_key(row)] = values
        return merge_map


class RenameFilter(AbstractStreamingFilter):
    """
    Composable filter class to rename columns in a HXL dataset.

    This is the class supporting the hxlrename command-line utility.

    Usage:

    <pre>
    hxl.data(url).rename_columns('#foo:New header#bar')
    </pre>
    """

    def __init__(self, source, rename=[]):
        """
        Constructor
        @param source the Dataset for the data.
        @param rename_map map of tags to rename
        """
        super(RenameFilter, self).__init__(source)
        if isinstance(rename, six.string_types):
            rename = [rename]
        self.rename = [RenameFilter.parse_rename(spec) for spec in rename]

    def filter_columns(self):
        """Rename requested columns."""
        return [self._rename_column(column) for column in self.source.columns]

    def filter_row(self, row):
        """Row will be rebuilt with proper columns."""
        return copy.copy(row.values)

    def _rename_column(self, column):
        """Rename a column if requested."""
        for spec in self.rename:
            if spec[0].match(column):
                new_column = copy.deepcopy(spec[1])
                if new_column.header is None:
                    new_column.header = column.header
                return new_column
        return copy.deepcopy(column)
    RENAME_PATTERN = r'^\s*#?({token}(?:\s*[+-]{token})*):(?:([^#]*)#)?({token}(?:\s*[+]{token})*)\s*$'.format(token=hxl.common.TOKEN)

    @staticmethod
    def parse_rename(s):
        """Parse a rename specification from the parameters."""
        if isinstance(s, six.string_types):
            result = re.match(RenameFilter.RENAME_PATTERN, s)
            if result:
                pattern = hxl.model.TagPattern.parse(result.group(1))
                column = hxl.model.Column.parse('#' + result.group(3), header=result.group(2), use_exception=True)
                return (pattern, column)
            else:
                raise HXLFilterException("Bad rename expression: " + s)
        else:
            return s


class ReplaceDataFilter(AbstractStreamingFilter):
    """
    Composable filter class to replace values in a HXL dataset.

    This is the class supporting the hxlreplace console script.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.
    """

    def __init__(self, source, replacements):
        """
        Constructor
        @param source the HXL data source
        @param original a string or regular expression to replace (string must match the whole value, not just part)
        @param replacement the replacement string (if using a regex, may contain substitution patterns)
        @param patterns (optional) a tag pattern or list of tag patterns - if present, constrain replacement to just these columns
        @param use_regex (optional) if True, then original is a regular expression rather than a string constant
        """
        super(ReplaceDataFilter, self).__init__(source)
        self.replacements = replacements
        if isinstance(self.replacements, ReplaceDataFilter.Replacement):
            self.replacements = [self.replacements]

    def filter_row(self, row):
        values = copy.copy(row.values)
        for index, value in enumerate(values):
            for replacement in self.replacements:
                value = replacement.sub(row.columns[index], value)
                values[index] = value
        return values


    class Replacement:
        """Replacement specification."""

        def __init__(self, original, replacement, pattern=None, is_regex=False):
            """
            @param original a string (case- and space-insensitive) or regular expression (sensitive) to replace
            @param replacement the replacement string or regular expression substitution
            @param pattern (optional) a tag pattern to limit the replacement to specific columns
            @param is_regex (optional) True to use regular-expression processing (defaults to False)
            """
            self.original = original
            self.replacement = replacement
            if pattern:
                self.pattern = hxl.model.TagPattern.parse(pattern)
            else:
                self.pattern = None
            self.is_regex = is_regex
            if not self.is_regex:
                self.original = hxl.common.normalise_string(self.original)

        def sub(self, column, value):
            """
            Substitute inside the value, if appropriate.
            @param column the column definition
            @param value the cell value
            @return the value, possibly changed
            """
            if self.pattern and not self.pattern.match(column):
                return value
            elif self.is_regex:
                return re.sub(self.original, self.replacement, value)
            elif self.original == hxl.common.normalise_string(value):
                return self.replacement
            else:
                return value

        @staticmethod
        def parse_map(source):
            """Parse a substitution map."""
            replacements = []
            for row in source:
                if row.get('#x_pattern'):
                    replacements.append(
                        ReplaceDataFilter.Replacement(
                            row.get('#x_pattern'), row.get('#x_substitution'),
                            row.get('#x_tag'), row.get('#x_regex')
                        ))
            return replacements

        
class RowFilter(AbstractStreamingFilter):
    """
    Composable filter class to select rows from a HXL dataset.

    Usage:

    <pre>
    # whitelist
    hxl.data(url).with_rows('org=OXFAM')

    # blacklist
    hxl.data(url).without_rows('org=OXFAM')
    </pre>
    """

    def __init__(self, source, queries=[], reverse=False):
        """
        Constructor
        @param source the HXL data source
        @param queries a series for parsed queries
        @param reverse True to reverse the sense of the select
        """
        super(RowFilter, self).__init__(source)
        if not hasattr(queries, '__len__') or isinstance(queries, six.string_types):
            # make a list if needed
            queries = [queries]
        self.queries = [hxl.model.RowQuery.parse(query) for query in queries]
        self.reverse = reverse

    def filter_row(self, row):
        if self.match_row(row):
            return row.values
        else:
            return None

    def match_row(self, row):
        """Check if any of the queries matches the row (implied OR)."""
        for query in self.queries:
            if query.match_row(row):
                return not self.reverse
        return self.reverse

    
class SortFilter(hxl.model.Dataset):
    """
    Composable filter class to sort a HXL dataset.

    This is the class supporting the hxlsort command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.

    Usage:

    <pre>
    source = HXLReader(sys.stdin)
    filter = SortFilter(source, tags=[TagPattern.parse('#sector'), TagPattern.parse('#org'), TagPattern.parse('#adm1']))
    write_hxl(sys.stdout, filter)
    </pre>
    """

    def __init__(self, source, tags=[], reverse=False):
        """
        @param source a HXL data source
        @param tags list of TagPattern objects for sorting
        @param reverse True to reverse the sort order
        """
        self.source = source
        self.sort_tags = hxl.model.TagPattern.parse_list(tags)
        self.reverse = reverse
        self._iter = None

    @property
    def columns(self):
        """
        Return the same columns as the source.
        """
        return self.source.columns

    def __iter__(self):
        """
        Sort the dataset first, then return it row by row.
        """

        # Closures
        def make_key(row):
            """Closure: use the requested tags as keys (if provided). """

            key = []

            def add_value(tag, value):
                """
                Split each value into a numeric and non-numeric representation.
                Non-numeric values sort as infinity (after all numeric ones).
                """
                norm = hxl.common.normalise_string(value)
                if tag == '#date':
                    key.append(float('inf'))
                    key.append(dateutil.parser.parse(norm).strftime('%Y-%m-%d'))
                else:
                    try:
                        key.append(float(norm))
                        key.append(norm)
                    except:
                        key.append(float('inf'))
                        key.append(norm)

            if self.sort_tags:
                # Specific sort tags chosen
                for pattern in self.sort_tags:
                    add_value(pattern.tag, pattern.get_value(row))
            else:
                # Sort everything, left to right
                for index, value in enumerate(row):
                    add_value(row.columns[index].tag, value)

            return tuple(key)

        # Main method
        return iter(sorted(self.source, key=make_key, reverse=self.reverse))


class ValidateFilter(hxl.model.Dataset):
    """Composable filter class to validate a HXL dataset against a schema.

    This is the class supporting the hxlvalidate command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, singled-threaded processing pipeline.

    Usage:

    <pre>
    source = HXLReader(sys.stdin)
    schema = hxl.schema('my-schema.csv')
    filter = ValidateFilter(source, schema)
    write_hxl(sys.stdout, filter)
    </pre>
    """

    def __init__(self, source, schema, show_all=False):
        """
        @param source a HXL data source
        @param schema a Schema object
        @param show_all boolean flag to report all lines (including those without errors).
        """
        self.source = source
        self.schema = schema
        self.show_all = show_all
        self._saved_columns = None

    @property
    def columns(self):
        """
        Add columns for the error reporting.
        """
        if self._saved_columns is None:
            # append error columns
            err_col = hxl.model.Column(tag='#x_errors', header='Error messages')
            tag_col = hxl.model.Column(tag='#x_tags', header='Error tag')
            row_col = hxl.model.Column(tag='#x_rows', header='Error row number (source)')
            col_col = hxl.model.Column(tag='#x_cols', header='Error column number (source)')
            self._saved_columns = self.source.columns + [err_col, tag_col, row_col, col_col]
        return self._saved_columns

    def __iter__(self):
        return ValidateFilter.Iterator(self)

    class Iterator:

        def __init__(self, outer):
            self.outer = outer
            self.iterator = iter(outer.source)

        def __iter__(self):
            return self

        def __next__(self):
            """
            Report rows with error information.
            """
            validation_errors = []
            def callback(error):
                """
                Collect validation errors
                """
                validation_errors.append(error)
            self.outer.schema.callback = callback

            """
            Read rows until we find an error (unless we're printing all rows)
            """
            row = next(self.iterator)
            while row:
                if not self.outer.schema.validate_row(row) or self.outer.show_all:
                    # append error data to row
                    error_row = copy.deepcopy(row)
                    messages = "\n".join(map(lambda e: e.message, validation_errors))
                    tags = "\n".join(map(lambda e: str(e.rule.tag_pattern) if e.rule else '', validation_errors))
                    rows = "\n".join(map(lambda e: str(e.row.row_number) if e.row else '', validation_errors))
                    error_row.columns = self.outer.columns
                    error_row.values = error_row.values + [messages, tags, rows]
                    return error_row
                else:
                    row = next(self.iterator)

        next = __next__

<commit_msg>Removed ValidateFilter (it was a silly idea).<commit_after>"""
Filters for the Humanitarian Exchange Language (HXL) v1.0
<NAME>
Started October 2014

License: Public Domain
Documentation: https://github.com/HXLStandard/libhxl-python/wiki
"""

import sys, re, six, abc, copy
import dateutil.parser

import hxl


#
# Filter-specific exception
#
class HXLFilterException(hxl.common.HXLException):
    pass

#
# Base class for filters

class AbstractFilter(hxl.model.Dataset):
    """
    Abstract base class for composable filters.

    This class stores the upstream source, and provides a
    filter_columns() method that child classes can implement. It will
    be called precisely once for each instantiation, giving the child
    a chance to provide a different set of columns than those in the
    source.
    """

    __metaclass__ = abc.ABCMeta

    def __init__(self, source):
        """
        Construct a new abstract filter.
        @param source the source dataset
        """
        self.source = source
        self.filtered_columns = None

    @property
    def columns(self):
        if self.filtered_columns is None:
            self.filtered_columns = self.filter_columns()
        return self.filtered_columns

    def filter_columns(self):
        """
        Return a new list of columns for the filtered dataset.
        @return a list of hxl.model.Column objects
        """
        return self.source.columns

    
class AbstractStreamingFilter(AbstractFilter):
    """
    Abstract base class for streaming filters.

    A streaming filter processes one row at a time.  It can skip rows,
    but it never reorders them.  Child classes will implement the
    filter_columns() method from the AbstractFilter class, as well as
    the filter_row(row) method from this class.
    """

    __metaclass__ = abc.ABCMeta

    def __init__(self, source):
        super(AbstractStreamingFilter, self).__init__(source)

    @abc.abstractmethod
    def filter_row(self, row):
        """
        Filter a single row of data.
        A return value of None will cause the row to be skipped.

        @param row A hxl.model.Row object to filter.
        @return An array of new values (not a Row object) or None to skip the row.
        """
        return row.values

    def __iter__(self):
        return AbstractStreamingFilter.Iterator(self)

    class Iterator:
        """
        Iterator to return the filtered rows.
        """

        def __init__(self, outer):
            self.outer = outer
            self.source_iter = iter(self.outer.source)
            self.row_number = -1

        def __iter__(self):
            return self

        def __next__(self):
            # call this here, in case it caches any useful information
            columns = self.outer.columns
            for row in self.source_iter:
                # get a new list of filtered values
                values = self.outer.filter_row(row)
                if values is not None:
                    # keep looping if filter_row(row) returned None
                    self.row_number += 1
                    # create a new Row object
                    return hxl.model.Row(columns, values, self.row_number)
            # if we've finished the iteration, then we're out of rows, so stop
            raise StopIteration()

        next = __next__


class AbstractCachingFilter(AbstractFilter):
    """
    Abstract base class for caching filters.

    A caching filter reads all of the input data first,
    then returns it, possibly in a different order
    or even completely transformed.
    """

    __metaclass__ = abc.ABCMeta

    def __init__(self, source):
        super(AbstractCachingFilter, self).__init__(source)

    def filter_rows(self):
        return self.source.values

    def __iter__(self):
        return AbstractCachingFilter.Iterator(self)

    class Iterator:

        def __init__(self, outer):
            self.outer = outer
            self.values_iter = None
            self.row_number = -1

        def __iter__(self):
            return self

        def __next__(self):
            if self.values_iter is None:
                self.values_iter = iter(outer.filter_rows())
            self.row_number += 1
            return hxl.model.Row(outer.columns, next(self.values_iter), self.row_number)
        
        next = __next__


#
# Filter classes
#

class AddColumnsFilter(AbstractStreamingFilter):
    """
    Composable filter class to add constant values to every row of a HXL dataset.

    This is the class supporting the hxladd command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.
    """

    def __init__(self, source, specs, before=False):
        """
        @param source a HXL data source
        @param specs a sequence of pairs of Column objects and constant values
        @param before True to add new columns before existing ones
        """
        super(AddColumnsFilter, self).__init__(source)
        if isinstance(specs, six.string_types):
            specs = [specs]
        self.specs = [AddColumnsFilter.parse_spec(spec) for spec in specs]
        self.before = before
        self.const_values = None

    def filter_columns(self):
        new_columns = [spec[0] for spec in self.specs]
        if self.before:
            new_columns = new_columns + self.source.columns
        else:
            new_columns = self.source.columns + new_columns
        self.const_values = [spec[1] for spec in self.specs]
        return new_columns

    def filter_row(self, row):
        values = copy.copy(row.values)
        if self.before:
            return self.const_values + values
        else:
            return values + self.const_values

    VALUE_PATTERN = r'^\s*(?:([^#]*)#)?({token}(?:\s*\+{token})*)=(.*)\s*$'.format(token=hxl.common.TOKEN)

    @staticmethod
    def parse_spec(spec):
        if not isinstance(spec, six.string_types):
            return spec
        result = re.match(AddColumnsFilter.VALUE_PATTERN, spec)
        if result:
            header = result.group(1)
            tag = '#' + result.group(2)
            value = result.group(3)
            return (hxl.model.Column(tag=tag, header=header), value)
        else:
            raise HXLFilterException("Badly formatted new-column spec: " + spec)


class AppendFilter(hxl.model.Dataset):
    """Composable filter class to concatenate two datasets."""

    def __init__(self, source, append_source, add_columns=True):
        """
        Constructor
        @param source the HXL data source
        @param append_source the HXL source to append
        @param add_columns flag for adding extra columns in append_source but not source (default True)
        """
        self.source = source
        self.append_source = append_source
        self.add_columns = add_columns
        self._saved_columns = None
        self._column_positions = None
        self._template_row = None

    @property
    def columns(self):
        if self._saved_columns is None:
            # Merge the columns from the second source into the first,
            # appending if necessary (display tag is the key)
            saved_columns = list(self.source.columns)
            column_positions = {}
            original_tags = self.source.display_tags

            # see if there's a corresponding column in the source
            for i, column in enumerate(self.append_source.columns):
                for j, tag in enumerate(original_tags):
                    if tag and (column.display_tag == tag):
                        # yes, there is one; clear it, so it's not reused
                        original_tags[j] = None
                        column_positions[i] = j
                        break
                else:
                    # no -- we need to add a new column
                    if self.add_columns:
                        column_positions[i] = len(saved_columns)
                        saved_columns.append(copy.deepcopy(column))
                    else:
                        column_positions[i] = None
            self._column_positions = column_positions
            self._saved_columns = saved_columns

            # make an empty template for each row
            self._template_row = [''] * len(saved_columns)

        # return the (usually cached) columns
        return self._saved_columns

    def __iter__(self):
        self.columns # make sure this is triggered first
        return AppendFilter.Iterator(self)

    class Iterator:
        """Custom iterator to return the contents of both sources, in sequence."""

        def __init__(self, outer):
            self.outer = outer
            self.source_iter = iter(outer.source)
            self.append_iter = iter(outer.append_source)

        def __iter__(self):
            return self

        def __next__(self):

            # Read from the original source first
            if self.source_iter is not None:
                try:
                    row_in = next(self.source_iter)
                    row_out = copy.deepcopy(row_in)
                    row_out.values = copy.deepcopy(self.outer._template_row)
                    for i, value in enumerate(row_in):
                        row_out.values[i] = row_in.values[i]
                    return row_out
                except StopIteration:
                    # don't let the end of the first source finish the iteration
                    self.source_iter = None

            # Fall through to the append source
            row_in = next(self.append_iter)
            row_out = copy.deepcopy(row_in)
            row_out.values = copy.deepcopy(self.outer._template_row)
            for i, value in enumerate(row_in):
                pos = self.outer._column_positions[i]
                if pos is not None:
                    row_out.values[pos] = value
            return row_out

        next = __next__

        
class CacheFilter(hxl.model.Dataset):
    """Composable filter class to cache HXL data in memory."""

    def __init__(self, source, max_rows=None):
        """
        Constructor
        @param max_rows If >0, maximum number of rows to cache.
        """
        self.source = source
        self.max_rows = max_rows
        self.cached_columns = copy.deepcopy(source.columns)
        self.cached_rows = [copy.deepcopy(row) for row in source]
        self.overflow = False

    @property
    def columns(self):
        return self.cached_columns

    def __iter__(self):
        return iter(self.cached_rows)

    def _load(self):
        if self.cached_rows is None:
            self.cached_rows = []
            self.cached_columns = copy.deepcopy(self.source.columns)
            row_count = 0
            for row in self.source:
                row_count += 1
                if self.max_rows > 1 and row_count >= self.max_rows:
                    self.overflow = True
                    break
                else:
                    self.cached_rows.append(copy.deepcopy(row))


class CleanDataFilter(AbstractStreamingFilter):
    """
    Filter for cleaning values in HXL data.
    Can normalise whitespace, convert to upper/lowercase, and fix dates and numbers.
    TODO: clean up lat/lon coordinates
    """

    def __init__(self, source, whitespace=False, upper=[], lower=[], date=[], number=[]):
        """
        Construct a new data-cleaning filter.
        @param source the HXLDataSource
        @param whitespace list of TagPatterns for normalising whitespace, or True to normalise all.
        @param upper list of TagPatterns for converting to uppercase, or True to convert all.
        @param lower list of TagPatterns for converting to lowercase, or True to convert all.
        @param lower list of TagPatterns for normalising dates, or True to normalise all ending in "_date"
        @param lower list of TagPatterns for normalising numbers, or True to normalise all ending in "_num"
        """
        super(CleanDataFilter, self).__init__(source)
        self.whitespace = whitespace
        self.upper = upper
        self.lower = lower
        self.date = date
        self.number = number

    def filter_row(self, row):
        """Clean up values and pass on the row data."""
        columns = self.columns
        values = copy.copy(row.values)
        for i in range(min(len(values), len(columns))):
            values[i] = self._clean_value(values[i], columns[i])
        return values

    def _clean_value(self, value, column):
        """Clean a single HXL value."""

        value = str(value)

        # Whitespace (-w or -W)
        if self._match_patterns(self.whitespace, column):
            value = re.sub('^\s+', '', value)
            value = re.sub('\s+$', '', value)
            value = re.sub('\s+', ' ', value)

        # Uppercase (-u)
        if self._match_patterns(self.upper, column):
            if sys.version_info[0] > 2:
                value = value.upper()
            else:
                value = value.decode('utf8').upper().encode('utf8')

        # Lowercase (-l)
        if self._match_patterns(self.lower, column):
            if sys.version_info[0] > 2:
                value = value.lower()
            else:
                value = value.decode('utf8').lower().encode('utf8')

        # Date (-d or -D)
        if self._match_patterns(self.date, column, '_date'):
            if value:
                value = dateutil.parser.parse(value).strftime('%Y-%m-%d')

        # Number (-n or -N)
        if self._match_patterns(self.number, column, '_num') and re.match('\d', value):
            if value:
                value = re.sub('[^\d.]', '', value)
                value = re.sub('^0+', '', value)
                value = re.sub('(\..*)0+$', '\g<1>', value)
                value = re.sub('\.$', '', value)

        return value

    def _match_patterns(self, patterns, column, extension=None):
        """Test if a column matches a list of patterns."""
        if not patterns:
            return False
        elif patterns is True:
            # if there's an extension specific like "_date", must match it
            return (column.tag and (not extension or column.tag.endswith(extension)))
        else:
            for pattern in patterns:
                if pattern.match(column):
                    return True
            return False


class ColumnFilter(AbstractStreamingFilter):
    """
    Composable filter class to filter columns in a HXL dataset.

    Usage:

    <pre>
    # blacklist columns
    hxl.data(url).without_columns('contact+email')

    # whitelist columns
    hxl.data(url).with_columns(['org', 'sector', 'adm1'])
    </pre>
    """

    def __init__(self, source, include_tags=[], exclude_tags=[]):
        """
        @param source a HXL data source
        @param include_tags a whitelist of TagPattern objects to include
        @param exclude_tags a blacklist of TagPattern objects to exclude
        """
        super(ColumnFilter, self).__init__(source)
        self.include_tags = hxl.model.TagPattern.parse_list(include_tags)
        self.exclude_tags = hxl.model.TagPattern.parse_list(exclude_tags)
        self.indices = [] # saved indices for columns to include

    def filter_columns(self):
        """
        Remove any columns in the blacklist or not in the whitelist.
        Save the indices in self.cached_indices for row filtering.
        """
        columns_in = self.source.columns
        columns_out = []
        for i in range(len(columns_in)):
            if self._test_column(columns_in[i]):
                columns_out.append(copy.deepcopy(columns_in[i]))
                self.indices.append(i) # save index to avoid retesting for data
        return columns_out

    def filter_row(self, row):
        """Remove values from a row for any column that's been removed."""
        values = []
        for i in self.indices:
            values.append(row.values[i])
        return values

    def _test_column(self, column):
        """
        Test whether a column should be included in the output.
        If there is a whitelist, it must be in the whitelist; if there is a blacklist, it must not be in the blacklist.
        """
        if self.exclude_tags:
            # blacklist
            for pattern in self.exclude_tags:
                if pattern.match(column):
                    # fail as soon as we match an excluded pattern
                    return False

        if self.include_tags:
            # whitelist
            for pattern in self.include_tags:
                if pattern.match(column):
                    # succeed as soon as we match an included pattern
                    return True
            # fail if there was a whitelist and we didn't match
            return False
        else:
            # no whitelist
            return True


class CountFilter(hxl.model.Dataset):
    """
    Composable filter class to aggregate rows in a HXL dataset.

    This is the class supporting the hxlcount command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.

    WARNING: this filter reads the entire source dataset before
    producing output, and may need to hold a large amount of data in
    memory, depending on the number of unique combinations counted.

    Usage:

    <pre>
    filter = CountFilter(source, patterns=['#org', '#sector'])
    </pre>

    or

    <pre>
    filter = source.count(['#org', '#sector'])
    </pre>
    """

    def __init__(self, source, patterns, aggregate_pattern=None):
        """
        Constructor
        @param source the HXL data source
        @param patterns a list of strings or TagPattern objects that form a unique key together
        @param aggregate_pattern an optional tag pattern calculating numeric aggregate values.
        """
        self.source = source
        self.patterns = hxl.model.TagPattern.parse_list(patterns)
        self.aggregate_pattern = hxl.model.TagPattern.parse(aggregate_pattern) if aggregate_pattern else None
        self._saved_columns = None

    @property
    def columns(self):
        """
        @return the column definitions used in the aggregation report
        """
        if self._saved_columns is None:
            cols = []
            for pattern in self.patterns:
                column = pattern.find_column(self.source.columns)
                if column is not None:
                    header = column.header
                else:
                    header = None
                cols.append(hxl.model.Column(tag=pattern.tag, attributes=pattern.include_attributes, header=header))
            cols.append(hxl.model.Column.parse('#meta+count', 'Count'))
            if self.aggregate_pattern is not None:
                cols.append(hxl.model.Column.parse('#meta+sum', header='Sum'))
                cols.append(hxl.model.Column.parse('#meta+average', header='Average (mean)'))
                cols.append(hxl.model.Column.parse('#meta+min', header='Minimum value'))
                cols.append(hxl.model.Column.parse('#meta+max', header='Maximum value'))
            self._saved_columns = cols
        return self._saved_columns

    def __iter__(self):
        return CountFilter.Iterator(self)

    class Iterator:

        def __init__(self, outer):
            self.outer = outer
            self.iterator = iter(outer.source)
            self.aggregate_iter = None

        def __iter__(self):
            return self

        def __next__(self):
            """
            @return the next row of aggregated data.
            """
            if self.aggregate_iter is None:
                self._aggregate()
            # Write the stats, sorted in value order
            aggregate = next(self.aggregate_iter)
            values = list(aggregate[0])
            values.append(aggregate[1].count)
            if self.outer.aggregate_pattern:
                if aggregate[1].seen_numbers:
                    values.append(aggregate[1].sum)
                    values.append(aggregate[1].average)
                    values.append(aggregate[1].min)
                    values.append(aggregate[1].max)
                else:
                    values = values + ([''] * 4)

            row = hxl.model.Row(self.outer.columns)
            row.values = values
            return row

        next = __next__

        def _aggregate(self):
            """
            Read the entire source dataset and produce saved aggregate data.
            """
            aggregators = {}
            for row in self.iterator:
                values = [str(pattern.get_value(row)) for pattern in self.outer.patterns]
                if values:
                    key = tuple(values)
                    if not key in aggregators:
                        aggregators[key] = CountFilter.Aggregator(self.outer.aggregate_pattern)
                    aggregators[key].add(row)
            self.aggregate_iter = iter(sorted(aggregators.items()))

    class Aggregator(object):
        """
        Class to collect aggregates for a single combination.

        Currently calculates count, sum, average, min, and max
        """

        def __init__(self, pattern):
            """
            Constructor
            @param pattern the HXL tag being counted in the row.
            """
            self.pattern = pattern
            self.count = 0
            self.sum = 0.0
            self.average = 0.0
            self.min = None
            self.max = None
            self.seen_numbers = False

        def add(self, row):
            """
            Add a new row of data to the aggregator.
            """
            self.count += 1
            if self.pattern:
                value = self.pattern.get_value(row)
                if value:
                    try:
                        n = float(value)
                        self.sum += n
                        self.average = self.sum / self.count
                        if self.min is None or n < self.min:
                            self.min = n
                        if self.max is None or n > self.max:
                            self.max = n
                        self.seen_numbers = True
                    except:
                        pass


class DeduplicationFilter(AbstractStreamingFilter):
    """
    Composable filter to deduplicate a HXL dataset.

    Supports the hxldedup command-line script.

    TODO: add more-sophisticated matching, edit distance, etc.
    """

    def __init__(self, source, patterns=None):
        """
        Constructor
        @param source the upstream source dataset
        @param patterns if provided, a list of tag patterns for columns to use for uniqueness testing.
        """
        super(DeduplicationFilter, self).__init__(source)
        self.patterns = hxl.model.TagPattern.parse_list(patterns)
        self.seen_map = set() # row signatures that we've seen so far

    def filter_row(self, row):
        """Filter out any rows we've seen before."""
        if not row:
            return None
        key = self._make_key(row)
        if key in self.seen_map:
            return None
        # if we get to here, we haven't seen the row before
        self.seen_map.add(key)
        return copy.copy(row.values)

    def _is_key(self, col):
        """Check if a column is part of the key for deduplication."""
        if self.patterns:
            for pattern in self.patterns:
                if pattern.match(col):
                    return True
            return False
        else:
            return True

    def _make_key(self, row):
        """Create a tuple key for a row."""
        key = []
        for i, value in enumerate(row.values):
            if self._is_key(row.columns[i]):
                key.append(hxl.common.normalise_string(value))
        return tuple(key)

        
class MergeDataFilter(AbstractStreamingFilter):
    """
    Composable filter class to merge values from two HXL datasets.

    This is the class supporting the hxlmerge command-line utility.

    Warning: this filter may store a large amount of data in memory, depending on the merge.

    Usage:

    <pre>
    hxl.data(url).merge(merge_source=merge_source, keys='adm1_id', tags='adm1')
    </pre>
    """

    def __init__(self, source, merge_source, keys, tags, replace=False, overwrite=False):
        """
        Constructor.
        @param source the HXL data source.
        @param merge_source a second HXL data source to merge into the first.
        @param keys the shared key hashtags to use for the merge
        @param tags the tags to include from the second dataset
        """
        super(MergeDataFilter, self).__init__(source)
        self.merge_source = merge_source
        self.keys = hxl.model.TagPattern.parse_list(keys)
        self.merge_tags = hxl.model.TagPattern.parse_list(tags)
        self.replace = replace
        self.overwrite = overwrite

        self.merge_map = None

    def filter_columns(self):
        """Filter the columns to add newly-merged ones."""
        new_columns = []
        for pattern in self.merge_tags:
            if self.replace and pattern.find_column(self.source.columns):
                # will use existing column
                continue
            else:
                column = pattern.find_column(self.merge_source.columns)
                if column:
                    header = column.header
                else:
                    header = None
                new_columns.append(hxl.model.Column(tag=pattern.tag, attributes=pattern.include_attributes, header=header))
        return self.source.columns + new_columns

    def filter_row(self, row):
        """Set up a merged data row, replacing existing values if requested."""
        
        # First, check if we already have the merge map, and read it if not
        if self.merge_map is None:
            self.merge_map = self._read_merge()

        # Make a copy of the values
        values = copy.copy(row.values)

        # Look up the merge values, based on the --keys
        merge_values = self.merge_map.get(self._make_key(row), {})

        # Go through the merge tags
        for pattern in self.merge_tags:
            # Try to substitute in place?
            if self.replace:
                index = pattern.find_column_index(self.source.columns)
                if index is not None:
                    if self.overwrite or not row.values[index]:
                        values[index] = merge_values.get(pattern)
                    continue

            # otherwise, fall through
            values.append(merge_values.get(pattern, ''))
        return values

    def _make_key(self, row):
        """
        Make a tuple key for a row.
        """
        values = []
        for pattern in self.keys:
            values.append(hxl.common.normalise_string(pattern.get_value(row)))
        return tuple(values)

    def _read_merge(self):
        """
        Read the second (merging) dataset into memory.
        Stores only the values necessary for the merge.
        @return a map of merge values
        """
        merge_map = {}
        for row in self.merge_source:
            values = {}
            for pattern in self.merge_tags:
                values[pattern] = pattern.get_value(row)
            merge_map[self._make_key(row)] = values
        return merge_map


class RenameFilter(AbstractStreamingFilter):
    """
    Composable filter class to rename columns in a HXL dataset.

    This is the class supporting the hxlrename command-line utility.

    Usage:

    <pre>
    hxl.data(url).rename_columns('#foo:New header#bar')
    </pre>
    """

    def __init__(self, source, rename=[]):
        """
        Constructor
        @param source the Dataset for the data.
        @param rename_map map of tags to rename
        """
        super(RenameFilter, self).__init__(source)
        if isinstance(rename, six.string_types):
            rename = [rename]
        self.rename = [RenameFilter.parse_rename(spec) for spec in rename]

    def filter_columns(self):
        """Rename requested columns."""
        return [self._rename_column(column) for column in self.source.columns]

    def filter_row(self, row):
        """Row will be rebuilt with proper columns."""
        return copy.copy(row.values)

    def _rename_column(self, column):
        """Rename a column if requested."""
        for spec in self.rename:
            if spec[0].match(column):
                new_column = copy.deepcopy(spec[1])
                if new_column.header is None:
                    new_column.header = column.header
                return new_column
        return copy.deepcopy(column)
    RENAME_PATTERN = r'^\s*#?({token}(?:\s*[+-]{token})*):(?:([^#]*)#)?({token}(?:\s*[+]{token})*)\s*$'.format(token=hxl.common.TOKEN)

    @staticmethod
    def parse_rename(s):
        """Parse a rename specification from the parameters."""
        if isinstance(s, six.string_types):
            result = re.match(RenameFilter.RENAME_PATTERN, s)
            if result:
                pattern = hxl.model.TagPattern.parse(result.group(1))
                column = hxl.model.Column.parse('#' + result.group(3), header=result.group(2), use_exception=True)
                return (pattern, column)
            else:
                raise HXLFilterException("Bad rename expression: " + s)
        else:
            return s


class ReplaceDataFilter(AbstractStreamingFilter):
    """
    Composable filter class to replace values in a HXL dataset.

    This is the class supporting the hxlreplace console script.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.
    """

    def __init__(self, source, replacements):
        """
        Constructor
        @param source the HXL data source
        @param original a string or regular expression to replace (string must match the whole value, not just part)
        @param replacement the replacement string (if using a regex, may contain substitution patterns)
        @param patterns (optional) a tag pattern or list of tag patterns - if present, constrain replacement to just these columns
        @param use_regex (optional) if True, then original is a regular expression rather than a string constant
        """
        super(ReplaceDataFilter, self).__init__(source)
        self.replacements = replacements
        if isinstance(self.replacements, ReplaceDataFilter.Replacement):
            self.replacements = [self.replacements]

    def filter_row(self, row):
        values = copy.copy(row.values)
        for index, value in enumerate(values):
            for replacement in self.replacements:
                value = replacement.sub(row.columns[index], value)
                values[index] = value
        return values


    class Replacement:
        """Replacement specification."""

        def __init__(self, original, replacement, pattern=None, is_regex=False):
            """
            @param original a string (case- and space-insensitive) or regular expression (sensitive) to replace
            @param replacement the replacement string or regular expression substitution
            @param pattern (optional) a tag pattern to limit the replacement to specific columns
            @param is_regex (optional) True to use regular-expression processing (defaults to False)
            """
            self.original = original
            self.replacement = replacement
            if pattern:
                self.pattern = hxl.model.TagPattern.parse(pattern)
            else:
                self.pattern = None
            self.is_regex = is_regex
            if not self.is_regex:
                self.original = hxl.common.normalise_string(self.original)

        def sub(self, column, value):
            """
            Substitute inside the value, if appropriate.
            @param column the column definition
            @param value the cell value
            @return the value, possibly changed
            """
            if self.pattern and not self.pattern.match(column):
                return value
            elif self.is_regex:
                return re.sub(self.original, self.replacement, value)
            elif self.original == hxl.common.normalise_string(value):
                return self.replacement
            else:
                return value

        @staticmethod
        def parse_map(source):
            """Parse a substitution map."""
            replacements = []
            for row in source:
                if row.get('#x_pattern'):
                    replacements.append(
                        ReplaceDataFilter.Replacement(
                            row.get('#x_pattern'), row.get('#x_substitution'),
                            row.get('#x_tag'), row.get('#x_regex')
                        ))
            return replacements

        
class RowFilter(AbstractStreamingFilter):
    """
    Composable filter class to select rows from a HXL dataset.

    Usage:

    <pre>
    # whitelist
    hxl.data(url).with_rows('org=OXFAM')

    # blacklist
    hxl.data(url).without_rows('org=OXFAM')
    </pre>
    """

    def __init__(self, source, queries=[], reverse=False):
        """
        Constructor
        @param source the HXL data source
        @param queries a series for parsed queries
        @param reverse True to reverse the sense of the select
        """
        super(RowFilter, self).__init__(source)
        if not hasattr(queries, '__len__') or isinstance(queries, six.string_types):
            # make a list if needed
            queries = [queries]
        self.queries = [hxl.model.RowQuery.parse(query) for query in queries]
        self.reverse = reverse

    def filter_row(self, row):
        if self.match_row(row):
            return row.values
        else:
            return None

    def match_row(self, row):
        """Check if any of the queries matches the row (implied OR)."""
        for query in self.queries:
            if query.match_row(row):
                return not self.reverse
        return self.reverse

    
class SortFilter(hxl.model.Dataset):
    """
    Composable filter class to sort a HXL dataset.

    This is the class supporting the hxlsort command-line utility.

    Because this class is a {@link hxl.model.Dataset}, you can use
    it as the source to an instance of another filter class to build a
    dynamic, single-threaded processing pipeline.

    Usage:

    <pre>
    source = HXLReader(sys.stdin)
    filter = SortFilter(source, tags=[TagPattern.parse('#sector'), TagPattern.parse('#org'), TagPattern.parse('#adm1']))
    write_hxl(sys.stdout, filter)
    </pre>
    """

    def __init__(self, source, tags=[], reverse=False):
        """
        @param source a HXL data source
        @param tags list of TagPattern objects for sorting
        @param reverse True to reverse the sort order
        """
        self.source = source
        self.sort_tags = hxl.model.TagPattern.parse_list(tags)
        self.reverse = reverse
        self._iter = None

    @property
    def columns(self):
        """
        Return the same columns as the source.
        """
        return self.source.columns

    def __iter__(self):
        """
        Sort the dataset first, then return it row by row.
        """

        # Closures
        def make_key(row):
            """Closure: use the requested tags as keys (if provided). """

            key = []

            def add_value(tag, value):
                """
                Split each value into a numeric and non-numeric representation.
                Non-numeric values sort as infinity (after all numeric ones).
                """
                norm = hxl.common.normalise_string(value)
                if tag == '#date':
                    key.append(float('inf'))
                    key.append(dateutil.parser.parse(norm).strftime('%Y-%m-%d'))
                else:
                    try:
                        key.append(float(norm))
                        key.append(norm)
                    except:
                        key.append(float('inf'))
                        key.append(norm)

            if self.sort_tags:
                # Specific sort tags chosen
                for pattern in self.sort_tags:
                    add_value(pattern.tag, pattern.get_value(row))
            else:
                # Sort everything, left to right
                for index, value in enumerate(row):
                    add_value(row.columns[index].tag, value)

            return tuple(key)

        # Main method
        return iter(sorted(self.source, key=make_key, reverse=self.reverse))

# end

{'fraction_non_alphanumeric': 0.08729439191814532, 'fraction_numerical': 0.0013728791607304754, 'mean_word_length': 1.8744648374967425, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 49, 'https://': 2, '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': '3460521', 'n_tokens_mistral': 20767, 'n_tokens_neox': 19708, 'n_words': 6311}