stream.py

# -*- coding: utf-8 -*-
from __future__ importdivision
from __future__ importprint_function
from __future__ importabsolute_import
from __future__ importunicode_literals

importre
importsix
importgzip
importzipfile
importtempfile
importwarnings
fromcopyimportcopy
fromitertoolsimportchain
fromcollectionsimportdeque
from .loaders.streamimportStreamLoader
from . importexceptions
from . importhelpers
from . importconfig


# Module API

# TODO: merge pick/skip rows logic
classStream(object):
"""Stream of tabular data.

 This is the main `tabulator` class. It loads a data source, and allows you
 to stream its parsed contents.

 # Arguments

 source (str):
 Path to file as ``<scheme>\\://path/to/file.<format>``.
 If not explicitly set, the scheme (file, http, ...) and
 format (csv, xls, ...) are inferred from the source string.

 headers (Union[int, List[int], List[str]], optional):
 Either a row
 number or list of row numbers (in case of multi-line headers) to be
 considered as headers (rows start counting at 1), or the actual
 headers defined a list of strings. If not set, all rows will be
 treated as containing values.

 scheme (str, optional):
 Scheme for loading the file (file, http, ...).
 If not set, it'll be inferred from `source`.

 format (str, optional):
 File source's format (csv, xls, ...). If not
 set, it'll be inferred from `source`. inferred

 encoding (str, optional):
 Source encoding. If not set, it'll be inferred.

 compression (str, optional):
 Source file compression (zip, ...). If not set, it'll be inferred.

 pick_rows (List[Union[int, str, dict]], optional):
 The same as `skip_rows` but it's for picking rows instead of skipping.

 skip_rows (List[Union[int, str, dict]], optional):
 List of row numbers, strings and regex patterns as dicts to skip.
 If a string, it'll skip rows that their first cells begin with it e.g. '#' and '//'.
 To skip only completely blank rows use `{'type'\\: 'preset', 'value'\\: 'blank'}`
 To provide a regex pattern use `{'type'\\: 'regex', 'value'\\: '^#'}`
 For example\\: `skip_rows=[1, '# comment', {'type'\\: 'regex', 'value'\\: '^# (regex|comment)'}]`

 pick_fields (List[Union[int, str]], optional):
 When passed, ignores all columns with headers
 that the given list DOES NOT include

 skip_fields (List[Union[int, str]], optional):
 When passed, ignores all columns with headers
 that the given list includes. If it contains an empty string it will skip
 empty headers

 sample_size (int, optional):
 Controls the number of sample rows used to
 infer properties from the data (headers, encoding, etc.). Set to
 ``0`` to disable sampling, in which case nothing will be inferred
 from the data. Defaults to ``config.DEFAULT_SAMPLE_SIZE``.

 bytes_sample_size (int, optional):
 Same as `sample_size`, but instead
 of number of rows, controls number of bytes. Defaults to
 ``config.DEFAULT_BYTES_SAMPLE_SIZE``.

 allow_html (bool, optional):
 Allow the file source to be an HTML page.
 If False, raises ``exceptions.FormatError`` if the loaded file is
 an HTML page. Defaults to False.

 multiline_headers_joiner (str, optional):
 When passed, it's used to join multiline headers
 as `<passed-value>.join(header1_1, header1_2)`
 Defaults to ' ' (space).

 multiline_headers_duplicates (bool, optional):
 By default tabulator will exclude a cell of a miltilne header from joining
 if it's exactly the same as the previous seen value in this field.
 Enabling this option will force duplicates inclusion
 Defaults to False.

 hashing_algorithm (func, optional):
 It supports: md5, sha1, sha256, sha512
 Defaults to sha256

 force_strings (bool, optional):
 When True, casts all data to strings.
 Defaults to False.

 force_parse (bool, optional):
 When True, don't raise exceptions when
 parsing malformed rows, simply returning an empty value. Defaults
 to False.

 post_parse (List[function], optional):
 List of generator functions that
 receives a list of rows and headers, processes them, and yields
 them (or not). Useful to pre-process the data. Defaults to None.

 custom_loaders (dict, optional):
 Dictionary with keys as scheme names,
 and values as their respective ``Loader`` class implementations.
 Defaults to None.

 custom_parsers (dict, optional):
 Dictionary with keys as format names,
 and values as their respective ``Parser`` class implementations.
 Defaults to None.

 custom_writers (dict, optional):
 Dictionary with keys as writer format
 names, and values as their respective ``Writer`` class
 implementations. Defaults to None.

 **options (Any, optional): Extra options passed to the loaders and parsers.

 """

# Public

def__init__(self,
source,
headers=None,
scheme=None,
format=None,
encoding=None,
compression=None,
allow_html=False,
sample_size=config.DEFAULT_SAMPLE_SIZE,
bytes_sample_size=config.DEFAULT_BYTES_SAMPLE_SIZE,
ignore_blank_headers=False,
ignore_listed_headers=None,
ignore_not_listed_headers=None,
multiline_headers_joiner=' ',
multiline_headers_duplicates=False,
hashing_algorithm='sha256',
force_strings=False,
force_parse=False,
pick_columns=None,
skip_columns=None,
pick_fields=None,
skip_fields=None,
limit_fields=None,
offset_fields=None,
pick_rows=None,
skip_rows=None,
limit_rows=None,
offset_rows=None,
post_parse=[],
custom_loaders={},
custom_parsers={},
custom_writers={},
**options):

# Translate aliases
ifpick_fieldsisnotNone:
pick_columns=pick_fields
ifskip_fieldsisnotNone:
skip_columns=skip_fields
ifpick_columnsisnotNone:
ignore_not_listed_headers=pick_columns
ifskip_columnsisnotNone:
ignore_listed_headers=skip_columns
if''inskip_columns:
ignore_blank_headers=True

# Set headers
self.__headers=None
self.__headers_row=None
self.__headers_row_last=None
ifisinstance(headers, int):
self.__headers_row=headers
self.__headers_row_last=headers
elifisinstance(headers, (tuple, list)):
if (len(headers) ==2and
isinstance(headers[0], int) and
isinstance(headers[1], int)):
self.__headers_row=headers[0]
self.__headers_row_last=headers[1]
else:
self.__headers=list(headers)

# Set pick rows
self.__pick_rows=pick_rows
self.__pick_rows_by_numbers= []
self.__pick_rows_by_patterns= []
self.__pick_rows_by_comments= []
self.__pick_rows_by_presets= {}
fordirectiveincopy(pick_rowsor []):
ifisinstance(directive, int):
self.__pick_rows_by_numbers.append(directive)
elifisinstance(directive, dict):
ifdirective['type'] =='regex':
self.__pick_rows_by_patterns.append(re.compile(directive['value']))
elifdirective['type'] =='preset'anddirective['value'] =='blank':
self.__pick_rows_by_presets['blank'] =True
else:
raiseValueError('Not supported pick rows: %s'%directive)
else:
self.__pick_rows_by_comments.append(str(directive))

# Set skip rows
self.__skip_rows=skip_rows
self.__skip_rows_by_numbers= []
self.__skip_rows_by_patterns= []
self.__skip_rows_by_comments= []
self.__skip_rows_by_presets= {}
fordirectiveincopy(skip_rowsor []):
ifisinstance(directive, int):
self.__skip_rows_by_numbers.append(directive)
elifisinstance(directive, dict):
ifdirective['type'] =='regex':
self.__skip_rows_by_patterns.append(re.compile(directive['value']))
elifdirective['type'] =='preset'anddirective['value'] =='blank':
self.__skip_rows_by_presets['blank'] =True
else:
raiseValueError('Not supported skip rows: %s'%directive)
else:
self.__skip_rows_by_comments.append(str(directive))

# Support for pathlib.Path
ifhasattr(source, 'joinpath'):
source=str(source)

# Set attributes
self.__source=source
self.__scheme=scheme
self.__format=format
self.__encoding=encoding
self.__compression=compression
self.__allow_html=allow_html
self.__sample_size=sample_size
self.__bytes_sample_size=bytes_sample_size
self.__ignore_blank_headers=ignore_blank_headers
self.__ignore_listed_headers=ignore_listed_headers
self.__ignore_not_listed_headers=ignore_not_listed_headers
self.__multiline_headers_joiner=multiline_headers_joiner
self.__multiline_headers_duplicates=multiline_headers_duplicates
self.__ignored_headers_indexes= []
self.__hashing_algorithm=hashing_algorithm
self.__force_strings=force_strings
self.__force_parse=force_parse
self.__limit_fields=limit_fields
self.__offset_fields=offset_fields
self.__limit_rows=limit_rows
self.__offset_rows=offset_rows
self.__post_parse=copy(post_parse)
self.__custom_loaders=copy(custom_loaders)
self.__custom_parsers=copy(custom_parsers)
self.__custom_writers=copy(custom_writers)
self.__actual_scheme=scheme
self.__actual_format=format
self.__actual_encoding=encoding
self.__actual_compression=compression
self.__options=options
self.__sample_extended_rows= []
self.__field_positions=None
self.__loader=None
self.__parser=None
self.__row_number=0
self.__stats=None

def__enter__(self):
ifself.closed:
self.open()
returnself

def__exit__(self, type, value, traceback):
ifnotself.closed:
self.close()

def__iter__(self):
returnself.iter()

@property
defclosed(self):
"""Returns True if the underlying stream is closed, False otherwise.

 # Returns
 bool: whether closed

 """
returnnotself.__parserorself.__parser.closed

defopen(self):
"""Opens the stream for reading.

 # Raises:
 TabulatorException: if an error

 """
source=self.__source
options=copy(self.__options)

# Programming error assertions
assertself.__hashing_algorithminconfig.SUPPORTED_HASHING_ALGORITHMS

# Validate compression
ifself.__compression:
ifself.__compressionnotinconfig.SUPPORTED_COMPRESSION:
message='Not supported compression "%s"'%self.__compression
raiseexceptions.CompressionError(message)

# Get scheme and format if not already given
compression=None
ifself.__schemeisNoneorself.__formatisNone:
detected_scheme, detected_format=helpers.detect_scheme_and_format(source)
scheme=self.__schemeordetected_scheme
format=self.__formatordetected_format
# Get compression
fortypeinconfig.SUPPORTED_COMPRESSION:
ifself.__compression==typeordetected_format==type:
compression=type
else:
scheme=self.__scheme
format=self.__format

# Initiate loader
self.__loader=None
ifschemeisnotNone:
loader_class=self.__custom_loaders.get(scheme)
ifloader_classisNone:
ifschemenotinconfig.LOADERS:
message='Scheme "%s" is not supported'%scheme
raiseexceptions.SchemeError(message)
loader_path=config.LOADERS[scheme]
ifloader_path:
loader_class=helpers.import_attribute(loader_path)
ifloader_classisnotNone:
loader_options=helpers.extract_options(options, loader_class.options)
ifcompressionand'http_stream'inloader_class.options:
loader_options['http_stream'] =False
self.__loader=loader_class(
bytes_sample_size=self.__bytes_sample_size,
**loader_options)

# Zip compression
ifcompression=='zip'andsix.PY3:
source=self.__loader.load(source, mode='b')
withzipfile.ZipFile(source) asarchive:
name=archive.namelist()[0]
if'filename'inoptions.keys():
name=options['filename']
deloptions['filename']
witharchive.open(name) asfile:
source=tempfile.NamedTemporaryFile(suffix='.'+name)
forlineinfile:
source.write(line)
source.seek(0)
# We redefine loader/format/schema after decompression
self.__loader=StreamLoader(bytes_sample_size=self.__bytes_sample_size)
format=self.__formatorhelpers.detect_scheme_and_format(source.name)[1]
scheme='stream'

# Gzip compression
elifcompression=='gz'andsix.PY3:
name=''
ifisinstance(source, str):
name=source.replace('.gz', '')
source=gzip.open(self.__loader.load(source, mode='b'))
# We redefine loader/format/schema after decompression
self.__loader=StreamLoader(bytes_sample_size=self.__bytes_sample_size)
format=self.__formatorhelpers.detect_scheme_and_format(name)[1]
scheme='stream'

# Not supported compression
elifcompression:
message='Compression "%s" is not supported for your Python version'
raiseexceptions.TabulatorException(message%compression)

# Attach stats to the loader
ifgetattr(self.__loader, 'attach_stats', None):
self.__stats= {'size': 0, 'hash': '', 'hashing_algorithm': self.__hashing_algorithm}
getattr(self.__loader, 'attach_stats')(self.__stats)

# Initiate parser
parser_class=self.__custom_parsers.get(format)
ifparser_classisNone:
ifformatnotinconfig.PARSERS:
# If not existent it's a not-found error
# Next line will raise IOError/HTTPError
chars=self.__loader.load(source)
chars.close()
# Otherwise it's a format error
message='Format "%s" is not supported'%format
raiseexceptions.FormatError(message)
parser_class=helpers.import_attribute(config.PARSERS[format])
parser_options=helpers.extract_options(options, parser_class.options)
self.__parser=parser_class(self.__loader,
force_parse=self.__force_parse,
**parser_options)

# Bad options
ifoptions:
message='Not supported option(s) "%s" for scheme "%s" and format "%s"'
message=message% (', '.join(options), scheme, format)
warnings.warn(message, UserWarning)

# Open and setup
self.__parser.open(source, encoding=self.__encoding)
self.__extract_sample()
self.__extract_headers()
ifnotself.__allow_html:
self.__detect_html()

# Set scheme/format/encoding
self.__actual_scheme=scheme
self.__actual_format=format
self.__actual_encoding=self.__parser.encoding
self.__actual_compression=compression

returnself

defclose(self):
"""Closes the stream.
 """
self.__parser.close()
self.__row_number=0

defreset(self):
"""Resets the stream pointer to the beginning of the file.
 """
ifself.__row_number>self.__sample_size:
self.__stats= {'size': 0, 'hash': ''}
self.__parser.reset()
self.__extract_sample()
self.__extract_headers()
self.__row_number=0

@property
defsource(self):
"""Source

 # Returns
 any: stream source

 """
returnself.__source

@property
defheaders(self):
"""Headers

 # Returns
 str[]/None: headers if available

 """
returnself.__headers

@headers.setter
defheaders(self, headers):
"""Set headers

 # Arguments
 str[]: headers

 """
self.__headers=headers

@property
defscheme(self):
"""Path's scheme

 # Returns
 str: scheme

 """
returnself.__actual_schemeor'inline'

@property
defformat(self):
"""Path's format

 # Returns
 str: format

 """
returnself.__actual_formator'inline'

@property
defencoding(self):
"""Stream's encoding

 # Returns
 str: encoding

 """
returnself.__actual_encodingor'no'

@property
defcompression(self):
"""Stream's compression ("no" if no compression)

 # Returns
 str: compression

 """
returnself.__actual_compressionor'no'

@property
deffragment(self):
"""Path's fragment

 # Returns
 str: fragment

 """
ifself.__parser:
returngetattr(self.__parser, 'fragment', None)
returnNone

@property
defdialect(self):
"""Dialect (if available)

 # Returns
 dict/None: dialect

 """
ifself.__parser:
returngetattr(self.__parser, 'dialect', {})
returnNone

@property
defsize(self):
"""Returns the BYTE count of the read chunks if available

 # Returns
 int/None: BYTE count

 """
ifself.__stats:
returnself.__stats['size']

@property
defhash(self):
"""Returns the SHA256 (or according to the "hashing_algorithm" parameter)
 hash of the read chunks if available

 # Returns
 str/None: bytes hash

 """
ifself.__stats:
returnself.__stats['hash']

@property
defsample(self):
"""Returns the stream's rows used as sample.

 These sample rows are used internally to infer characteristics of the
 source file (e.g. encoding, headers, ...).

 # Returns
 list[]: sample

 """
sample= []
iterator=iter(self.__sample_extended_rows)
iterator=self.__apply_processors(iterator)
forrow_number, headers, rowiniterator:
sample.append(row)
returnsample

@property
deffield_positions(self):
ifself.__field_positionsisNone:
self.__field_positions= []
ifself.__headers:
size=len(self.__headers) +len(self.__ignored_headers_indexes)
forindexinrange(size):
ifindexnotinself.__ignored_headers_indexes:
self.__field_positions.append(index+1)
returnself.__field_positions

@property
defhashing_algorithm(self):
returnself.__hashing_algorithm

defiter(self, keyed=False, extended=False):
"""Iterate over the rows.

 Each row is returned in a format that depends on the arguments `keyed`
 and `extended`. By default, each row is returned as list of their
 values.

 # Arguments
 keyed (bool, optional):
 When True, each returned row will be a
 `dict` mapping the header name to its value in the current row.
 For example, `[{'name'\\: 'J Smith', 'value'\\: '10'}]`. Ignored if
 ``extended`` is True. Defaults to False.
 extended (bool, optional):
 When True, returns each row as a tuple
 with row number (starts at 1), list of headers, and list of row
 values. For example, `(1, ['name', 'value'], ['J Smith', '10'])`.
 Defaults to False.

 # Raises
 exceptions.TabulatorException: If the stream is closed.

 # Returns
 Iterator[Union[List[Any], Dict[str, Any], Tuple[int, List[str], List[Any]]]]:
 The row itself. The format depends on the values of `keyed` and
 `extended` arguments.

 """

# Error if closed
ifself.closed:
message='Stream is closed. Please call "stream.open()" first.'
raiseexceptions.TabulatorException(message)

# Create iterator
iterator=chain(
self.__sample_extended_rows,
self.__parser.extended_rows)
iterator=self.__apply_processors(iterator)

# Yield rows from iterator
try:
count=0
forrow_number, headers, rowiniterator:
ifrow_number>self.__row_number:
count+=1
ifself.__limit_rowsorself.__offset_rows:
offset=self.__offset_rowsor0
limit=self.__limit_rows+offsetifself.__limit_rowselseNone
ifoffsetandcount<=offset:
continue
iflimitandcount>limit:
break
self.__row_number=row_number
ifextended:
yield (row_number, headers, row)
elifkeyed:
yielddict(zip(headers, row))
else:
yieldrow
exceptUnicodeErroraserror:
message='Cannot parse the source "%s" using "%s" encoding at "%s"'
raiseexceptions.EncodingError(message% (self.__source, error.encoding, error.start))
exceptExceptionaserror:
raiseexceptions.SourceError(str(error))

defread(self, keyed=False, extended=False, limit=None):
"""Returns a list of rows.

 # Arguments
 keyed (bool, optional): See :func:`Stream.iter`.
 extended (bool, optional): See :func:`Stream.iter`.
 limit (int, optional):
 Number of rows to return. If None, returns all rows. Defaults to None.

 # Returns
 List[Union[List[Any], Dict[str, Any], Tuple[int, List[str], List[Any]]]]:
 The list of rows. The format depends on the values of `keyed`
 and `extended` arguments.
 """
result= []
rows=self.iter(keyed=keyed, extended=extended)
forcount, rowinenumerate(rows, start=1):
result.append(row)
ifcount==limit:
break
returnresult

defsave(self, target, format=None, encoding=None, **options):
"""Save stream to the local filesystem.

 # Arguments
 target (str): Path where to save the stream.
 format (str, optional):
 The format the stream will be saved as. If
 None, detects from the ``target`` path. Defaults to None.
 encoding (str, optional):
 Saved file encoding. Defaults to ``config.DEFAULT_ENCODING``.
 **options: Extra options passed to the writer.

 # Returns
 count (int?): Written rows count if available
 """

# Get encoding/format
ifencodingisNone:
encoding=config.DEFAULT_ENCODING
ifformatisNone:
_, format=helpers.detect_scheme_and_format(target)

# Prepare writer class
writer_class=self.__custom_writers.get(format)
ifwriter_classisNone:
ifformatnotinconfig.WRITERS:
message='Format "%s" is not supported'%format
raiseexceptions.FormatError(message)
writer_class=helpers.import_attribute(config.WRITERS[format])

# Prepare writer options
writer_options=helpers.extract_options(options, writer_class.options)
ifoptions:
message='Not supported options "%s" for format "%s"'
message=message% (', '.join(options), format)
raiseexceptions.TabulatorException(message)

# Write data to target
writer=writer_class(**writer_options)
returnwriter.write(self.iter(), target, headers=self.headers, encoding=encoding)

# Private

def__extract_sample(self):

# Sample is not requested
ifnotself.__sample_size:
return

# Extract sample rows
self.__sample_extended_rows= []
for_inrange(self.__sample_size):
try:
row_number, headers, row=next(self.__parser.extended_rows)
ifself.__headers_rowandself.__headers_row>=row_number:
ifself.__check_if_row_for_skipping(row_number, headers, row):
self.__headers_row+=1
self.__headers_row_last+=1
self.__sample_extended_rows.append((row_number, headers, row))
exceptStopIteration:
break
exceptUnicodeErroraserror:
message='Cannot parse the source "%s" using "%s" encoding at "%s"'
raiseexceptions.EncodingError(message% (self.__source, error.encoding, error.start))
exceptExceptionaserror:
raiseexceptions.SourceError(str(error))

def__extract_headers(self):

# Heders row is not set
ifnotself.__headers_row:
return

# Sample is too short
ifself.__headers_row>self.__sample_size:
message='Headers row (%s) can\'t be more than sample_size (%s)'
message=message% (self.__headers_row, self.__sample_size)
raiseexceptions.TabulatorException(message)

# Get headers from data
last_merged= {}
keyed_source=False
forrow_number, headers, rowinself.__sample_extended_rows:
keyed_source=keyed_sourceorheadersisnotNone
headers=headersifkeyed_sourceelserow
forindex, headerinenumerate(headers):
ifheaderisnotNone:
headers[index] =six.text_type(header).strip()
ifrow_number==self.__headers_row:
self.__headers=headers
last_merged= {index: headerforindex, headerinenumerate(headers)}
ifrow_number>self.__headers_row:
forindexinrange(0, len(self.__headers)):
iflen(headers) >indexandheaders[index] isnotNone:
ifnotself.__headers[index]:
self.__headers[index] =headers[index]
else:
if (self.__multiline_headers_duplicatesor
last_merged.get(index) !=headers[index]):
self.__headers[index] += (
self.__multiline_headers_joiner+headers[index])
last_merged[index] =headers[index]
ifrow_number==self.__headers_row_last:
break

# Ignore headers
if (self.__ignore_blank_headersor
self.__ignore_listed_headersisnotNoneor
self.__ignore_not_listed_headersisnotNone):
self.__ignored_headers_indexes= []
raw_headers, self.__headers=self.__headers, []
forindex, headerinlist(enumerate(raw_headers)):
ignore=False
# Ignore blank headers
ifheaderin ['', None]:
ignore=True
# Ignore listed headers
ifself.__ignore_listed_headersisnotNone:
if (headerinself.__ignore_listed_headersor
index+1inself.__ignore_listed_headers):
ignore=True
# Regex
foriteminself.__ignore_listed_headers:
ifisinstance(item, dict) anditem.get('type') =='regex':
ifbool(re.search(item['value'], header)):
ignore=True
# Ignore not-listed headers
ifself.__ignore_not_listed_headersisnotNone:
if (headernotinself.__ignore_not_listed_headersand
index+1notinself.__ignore_not_listed_headers):
ignore=True
# Regex
foriteminself.__ignore_not_listed_headers:
ifisinstance(item, dict) anditem.get('type') =='regex':
ifbool(re.search(item['value'], header)):
ignore=False
# Add to the list and skip
ifignore:
self.__ignored_headers_indexes.append(index)
continue
self.__headers.append(header)
self.__ignored_headers_indexes=list(sorted(self.__ignored_headers_indexes, reverse=True))

# Limit/offset fields
ifself.__limit_fieldsorself.__offset_fields:
ignore= []
headers= []
min=self.__offset_fieldsor0
max=self.__limit_fields+minifself.__limit_fieldselselen(self.__headers)
forposition, headerinenumerate(self.__headers, start=1):
ifposition<=min:
ignore.append(position-1)
continue
ifposition>max:
ignore.append(position-1)
continue
headers.append(header)
forindexinignore:
ifindexnotinself.__ignored_headers_indexes:
self.__ignored_headers_indexes.append(index)
self.__ignored_headers_indexes=list(sorted(self.__ignored_headers_indexes, reverse=True))
self.__headers=headers

# Remove headers from data
ifnotkeyed_source:
delself.__sample_extended_rows[:self.__headers_row_last]

# Stringify headers
ifisinstance(self.__headers, list):
str_headers= []
forheaderinself.__headers:
str_headers.append(six.text_type(header) ifheaderisnotNoneelse'')
self.__headers=str_headers

def__detect_html(self):

# Prepare text
text=''
forrow_number, headers, rowinself.__sample_extended_rows:
forvalueinrow:
ifisinstance(value, six.string_types):
text+=value

# Detect html content
html_source=helpers.detect_html(text)
ifhtml_source:
message='Format has been detected as HTML (not supported)'
raiseexceptions.FormatError(message)

def__apply_processors(self, iterator):

# Base processor
defbuiltin_processor(extended_rows):
forrow_number, headers, rowinextended_rows:

# Sync headers/row
ifheaders!=self.__headers:
ifheadersandself.__headers:
keyed_row=dict(zip(headers, row))
row= [keyed_row.get(header) forheaderinself.__headers]
elifself.__ignored_headers_indexes:
row= [valueforindex, valueinenumerate(row) ifindexnotinself.__ignored_headers_indexes]
headers=self.__headers

# Skip rows by numbers/comments
ifself.__check_if_row_for_skipping(row_number, headers, row):
continue

yield (row_number, headers, row)

# Skip nagative rows processor
defskip_negative_rows(extended_rows):
'''
 This processor will skip rows which counts from the end, e.g.
 -1: skip last row, -2: skip pre-last row, etc.
 Rows to skip are taken from Stream.__skip_rows_by_numbers
 '''
rows_to_skip= [nforninself.__skip_rows_by_numbersifn<0]
buffer_size=abs(min(rows_to_skip))
# collections.deque - takes O[1] time to push/pop values from any side.
buffer=deque()

# use buffer to save last rows
forrowinextended_rows:
buffer.append(row)
iflen(buffer) >buffer_size:
yieldbuffer.popleft()

# Now squeeze out the buffer
n=len(buffer)
fori, rowinenumerate(buffer):
ifi-nnotinrows_to_skip:
yieldrow

# Force values to strings processor
defforce_strings_processor(extended_rows):
forrow_number, headers, rowinextended_rows:
row=list(map(helpers.stringify_value, row))
yield (row_number, headers, row)

# Form a processors list
processors= [builtin_processor]
# if we have to delete some rows with negative index (counting from the end)
if [nforninself.__skip_rows_by_numbersifn<0]:
processors.insert(0, skip_negative_rows)
ifself.__post_parse:
processors+=self.__post_parse
ifself.__force_strings:
processors.append(force_strings_processor)

# Apply processors to iterator
forprocessorinprocessors:
iterator=processor(iterator)

returniterator

def__check_if_row_for_skipping(self, row_number, headers, row):

# Pick rows
ifself.__pick_rows:

# Skip by number
ifrow_numberinself.__pick_rows_by_numbers:
returnFalse

# Get first cell
cell=row[0] ifrowelseNone

# Handle blank cell/row
ifcellin [None, '']:
if''inself.__pick_rows_by_comments:
returnFalse
ifself.__pick_rows_by_presets.get('blank'):
ifnotlist(filter(lambdacell: cellnotin [None, ''], row)):
returnFalse
returnTrue

# Pick by pattern
forpatterninself.__pick_rows_by_patterns:
ifbool(pattern.search(cell)):
returnFalse

# Pick by comment
forcommentinfilter(None, self.__pick_rows_by_comments):
ifsix.text_type(cell).startswith(comment):
returnFalse

# Default
returnTrue

# Skip rows
ifself.__skip_rows:

# Skip by number
ifrow_numberinself.__skip_rows_by_numbers:
returnTrue

# Get first cell
cell=row[0] ifrowelseNone

# Handle blank cell/row
ifcellin [None, '']:
if''inself.__skip_rows_by_comments:
returnTrue
ifself.__skip_rows_by_presets.get('blank'):
ifnotlist(filter(lambdacell: cellnotin [None, ''], row)):
returnTrue
returnFalse

# Skip by pattern