mirror of
https://github.com/python/cpython.git
synced 2026-08-01 09:31:29 +08:00
b8862aeac3
Detect the padding by parsing the sample both with and without skipinitialspace and comparing the two readings, instead of testing whether every field following a delimiter starts with a space.
672 lines
26 KiB
Python
672 lines
26 KiB
Python
|
|
r"""
|
|
CSV parsing and writing.
|
|
|
|
This module provides classes that assist in the reading and writing
|
|
of Comma Separated Value (CSV) files, and implements the interface
|
|
described by PEP 305. Although many CSV files are simple to parse,
|
|
the format is not formally defined by a stable specification and
|
|
is subtle enough that parsing lines of a CSV file with something
|
|
like line.split(",") is bound to fail. The module supports three
|
|
basic APIs: reading, writing, and registration of dialects.
|
|
|
|
|
|
DIALECT REGISTRATION:
|
|
|
|
Readers and writers support a dialect argument, which is a convenient
|
|
handle on a group of settings. When the dialect argument is a string,
|
|
it identifies one of the dialects previously registered with the module.
|
|
If it is a class or instance, the attributes of the argument are used as
|
|
the settings for the reader or writer:
|
|
|
|
class excel:
|
|
delimiter = ','
|
|
quotechar = '"'
|
|
escapechar = None
|
|
doublequote = True
|
|
skipinitialspace = False
|
|
lineterminator = '\r\n'
|
|
quoting = QUOTE_MINIMAL
|
|
|
|
SETTINGS:
|
|
|
|
* quotechar - specifies a one-character string to use as the
|
|
quoting character. It defaults to '"'.
|
|
* delimiter - specifies a one-character string to use as the
|
|
field separator. It defaults to ','.
|
|
* skipinitialspace - specifies how to interpret spaces which
|
|
immediately follow a delimiter. It defaults to False, which
|
|
means that spaces immediately following a delimiter is part
|
|
of the following field.
|
|
* lineterminator - specifies the character sequence which should
|
|
terminate rows.
|
|
* quoting - controls when quotes should be generated by the writer.
|
|
It can take on any of the following module constants:
|
|
|
|
csv.QUOTE_MINIMAL means only when required, for example, when a
|
|
field contains either the quotechar or the delimiter
|
|
csv.QUOTE_ALL means that quotes are always placed around fields.
|
|
csv.QUOTE_NONNUMERIC means that quotes are always placed around
|
|
fields which do not parse as integers or floating-point
|
|
numbers.
|
|
csv.QUOTE_STRINGS means that quotes are always placed around
|
|
fields which are strings. Note that the Python value None
|
|
is not a string.
|
|
csv.QUOTE_NOTNULL means that quotes are only placed around fields
|
|
that are not the Python value None.
|
|
csv.QUOTE_NONE means that quotes are never placed around fields.
|
|
* escapechar - specifies a one-character string used to escape
|
|
the delimiter when quoting is set to QUOTE_NONE.
|
|
* doublequote - controls the handling of quotes inside fields. When
|
|
True, two consecutive quotes are interpreted as one during read,
|
|
and when writing, each quote character embedded in the data is
|
|
written as two quotes
|
|
"""
|
|
|
|
import types
|
|
from _csv import Error, writer, reader, register_dialect, \
|
|
unregister_dialect, get_dialect, list_dialects, \
|
|
field_size_limit, \
|
|
QUOTE_MINIMAL, QUOTE_ALL, QUOTE_NONNUMERIC, QUOTE_NONE, \
|
|
QUOTE_STRINGS, QUOTE_NOTNULL
|
|
from _csv import Dialect as _Dialect
|
|
|
|
from io import StringIO
|
|
|
|
__all__ = ["QUOTE_MINIMAL", "QUOTE_ALL", "QUOTE_NONNUMERIC", "QUOTE_NONE",
|
|
"QUOTE_STRINGS", "QUOTE_NOTNULL",
|
|
"Error", "Dialect", "excel", "excel_tab",
|
|
"field_size_limit", "reader", "writer",
|
|
"register_dialect", "get_dialect", "list_dialects", "Sniffer",
|
|
"unregister_dialect", "DictReader", "DictWriter",
|
|
"unix_dialect"]
|
|
|
|
|
|
class Dialect:
|
|
"""Describe a CSV dialect.
|
|
|
|
This must be subclassed (see csv.excel). Valid attributes are:
|
|
delimiter, quotechar, escapechar, doublequote, skipinitialspace,
|
|
lineterminator, quoting.
|
|
|
|
"""
|
|
_name = ""
|
|
_valid = False
|
|
# placeholders
|
|
delimiter = None
|
|
quotechar = None
|
|
escapechar = None
|
|
doublequote = None
|
|
skipinitialspace = None
|
|
lineterminator = None
|
|
quoting = None
|
|
|
|
def __init__(self):
|
|
if self.__class__ != Dialect:
|
|
self._valid = True
|
|
self._validate()
|
|
|
|
def _validate(self):
|
|
try:
|
|
_Dialect(self)
|
|
except TypeError as e:
|
|
# Re-raise to get a traceback showing more user code.
|
|
raise Error(str(e)) from None
|
|
|
|
class excel(Dialect):
|
|
"""Describe the usual properties of Excel-generated CSV files."""
|
|
delimiter = ','
|
|
quotechar = '"'
|
|
doublequote = True
|
|
skipinitialspace = False
|
|
lineterminator = '\r\n'
|
|
quoting = QUOTE_MINIMAL
|
|
register_dialect("excel", excel)
|
|
|
|
class excel_tab(excel):
|
|
"""Describe the usual properties of Excel-generated TAB-delimited files."""
|
|
delimiter = '\t'
|
|
register_dialect("excel-tab", excel_tab)
|
|
|
|
class unix_dialect(Dialect):
|
|
"""Describe the usual properties of Unix-generated CSV files."""
|
|
delimiter = ','
|
|
quotechar = '"'
|
|
doublequote = True
|
|
skipinitialspace = False
|
|
lineterminator = '\n'
|
|
quoting = QUOTE_ALL
|
|
register_dialect("unix", unix_dialect)
|
|
|
|
|
|
class DictReader:
|
|
def __init__(self, f, fieldnames=None, restkey=None, restval=None,
|
|
dialect="excel", *args, **kwds):
|
|
if fieldnames is not None and iter(fieldnames) is fieldnames:
|
|
fieldnames = list(fieldnames)
|
|
self._fieldnames = fieldnames # list of keys for the dict
|
|
self.restkey = restkey # key to catch long rows
|
|
self.restval = restval # default value for short rows
|
|
self.reader = reader(f, dialect, *args, **kwds)
|
|
self.dialect = dialect
|
|
self.line_num = 0
|
|
|
|
def __iter__(self):
|
|
return self
|
|
|
|
@property
|
|
def fieldnames(self):
|
|
if self._fieldnames is None:
|
|
try:
|
|
self._fieldnames = next(self.reader)
|
|
except StopIteration:
|
|
pass
|
|
self.line_num = self.reader.line_num
|
|
return self._fieldnames
|
|
|
|
@fieldnames.setter
|
|
def fieldnames(self, value):
|
|
self._fieldnames = value
|
|
|
|
def __next__(self):
|
|
if self.line_num == 0:
|
|
# Used only for its side effect.
|
|
self.fieldnames
|
|
row = next(self.reader)
|
|
self.line_num = self.reader.line_num
|
|
|
|
# unlike the basic reader, we prefer not to return blanks,
|
|
# because we will typically wind up with a dict full of None
|
|
# values
|
|
while row == []:
|
|
row = next(self.reader)
|
|
d = dict(zip(self.fieldnames, row))
|
|
lf = len(self.fieldnames)
|
|
lr = len(row)
|
|
if lf < lr:
|
|
d[self.restkey] = row[lf:]
|
|
elif lf > lr:
|
|
for key in self.fieldnames[lr:]:
|
|
d[key] = self.restval
|
|
return d
|
|
|
|
__class_getitem__ = classmethod(types.GenericAlias)
|
|
|
|
|
|
class DictWriter:
|
|
def __init__(self, f, fieldnames, restval="", extrasaction="raise",
|
|
dialect="excel", *args, **kwds):
|
|
if fieldnames is not None and iter(fieldnames) is fieldnames:
|
|
fieldnames = list(fieldnames)
|
|
self.fieldnames = fieldnames # list of keys for the dict
|
|
self.restval = restval # for writing short dicts
|
|
extrasaction = extrasaction.lower()
|
|
if extrasaction not in ("raise", "ignore"):
|
|
raise ValueError("extrasaction (%s) must be 'raise' or 'ignore'"
|
|
% extrasaction)
|
|
self.extrasaction = extrasaction
|
|
self.writer = writer(f, dialect, *args, **kwds)
|
|
|
|
def writeheader(self):
|
|
header = dict(zip(self.fieldnames, self.fieldnames))
|
|
return self.writerow(header)
|
|
|
|
def _dict_to_list(self, rowdict):
|
|
if self.extrasaction == "raise":
|
|
wrong_fields = rowdict.keys() - self.fieldnames
|
|
if wrong_fields:
|
|
raise ValueError("dict contains fields not in fieldnames: "
|
|
+ ", ".join([repr(x) for x in wrong_fields]))
|
|
return (rowdict.get(key, self.restval) for key in self.fieldnames)
|
|
|
|
def writerow(self, rowdict):
|
|
return self.writer.writerow(self._dict_to_list(rowdict))
|
|
|
|
def writerows(self, rowdicts):
|
|
return self.writer.writerows(map(self._dict_to_list, rowdicts))
|
|
|
|
__class_getitem__ = classmethod(types.GenericAlias)
|
|
|
|
|
|
class Sniffer:
|
|
'''
|
|
"Sniffs" the format of a CSV file (i.e. delimiter, quotechar)
|
|
Returns a Dialect object.
|
|
'''
|
|
# Characters which can be guessed as a delimiter if the delimiters
|
|
# argument is not specified.
|
|
_delimiter_candidates = [c for c in map(chr, range(128))
|
|
if not c.isalnum()]
|
|
|
|
def __init__(self):
|
|
# in case there is more than one possible delimiter
|
|
self.preferred = [',', '\t', ';', ' ', ':']
|
|
|
|
|
|
def sniff(self, sample, delimiters=None):
|
|
"""
|
|
Analyze the sample and return a Dialect subclass reflecting the
|
|
parameters found. If the optional delimiters parameter is
|
|
given, it is interpreted as a string containing possible valid
|
|
delimiter characters. Raises Error if the dialect cannot be
|
|
determined.
|
|
|
|
The dialect is guessed by parsing the sample with every
|
|
plausible combination of delimiter, quotechar and escapechar,
|
|
and choosing the combination which splits the sample into rows
|
|
with the most consistent number of fields.
|
|
|
|
A large sample is parsed incrementally: first only its
|
|
beginning, then, after eliminating the combinations which are
|
|
clearly worse than the leader, a several times larger part,
|
|
and so on.
|
|
|
|
If several combinations fit the sample equally well, the
|
|
delimiters listed in the preferred attribute are preferred, in
|
|
that order, no matter how many times each of them occurs.
|
|
"""
|
|
import re
|
|
from collections import defaultdict
|
|
|
|
if self._parses_as_single_column(sample):
|
|
# There is no delimiter to find; any combination could
|
|
# only find a bogus one inside the quoted fields.
|
|
raise Error("Could not determine delimiter")
|
|
|
|
chars = set(sample)
|
|
if delimiters is None:
|
|
delimiters = self._delimiter_candidates
|
|
delimiters = [d for d in delimiters
|
|
if d in chars and d not in '\r\n"\'\\']
|
|
# Combinations to try, numbered by preference for breaking
|
|
# ties. The unquoted combinations are parsed from the start;
|
|
# the rest stay dormant until the quote character occurs at
|
|
# the start of a field (see _split_dormant).
|
|
groups = defaultdict(list)
|
|
order = 0
|
|
# Only '\\' is tried as an escape character: others are not
|
|
# seen in the wild.
|
|
for escapechar in ('', '\\') if '\\' in chars else ('',):
|
|
for quotechar in '"', "'", '':
|
|
if quotechar and quotechar not in chars:
|
|
continue
|
|
for delimiter in delimiters:
|
|
groups[quotechar].append(
|
|
(order, delimiter, quotechar, escapechar))
|
|
order += 1
|
|
active = groups.pop('', [])
|
|
# Only non-empty groups were created; a plain dict cannot
|
|
# grow one by accident.
|
|
dormant = dict(groups)
|
|
|
|
# The initial window should cover the minimal number of rows
|
|
# required for elimination (see _eliminate_worse) at a typical
|
|
# line length, so that the first round can already eliminate.
|
|
window = 2000
|
|
# A line with its line break: '\r', '\n' or '\r\n' (the
|
|
# reader treats other line boundary characters as ordinary
|
|
# data, but does not support a bare '\r' inside a chunk).
|
|
# The \z alternative produces one final empty match.
|
|
line_re = re.compile(r'[^\r\n]*(?:\r\n|[\r\n]|\z)')
|
|
parsed = []
|
|
lines = []
|
|
first_round = True
|
|
while active or dormant:
|
|
end = min(window, len(sample))
|
|
part = sample[:end]
|
|
lines = line_re.findall(part)
|
|
del lines[-1]
|
|
cut = not part.endswith(('\r', '\n'))
|
|
for quotechar in list(dormant):
|
|
activated, still = self._split_dormant(
|
|
part, quotechar, dormant[quotechar])
|
|
active += activated
|
|
if still:
|
|
dormant[quotechar] = still
|
|
else:
|
|
del dormant[quotechar]
|
|
parsed = [(combo, self._try_dialect(lines, cut, *combo[1:]))
|
|
for combo in active]
|
|
if end == len(sample):
|
|
break
|
|
active = self._eliminate_worse(parsed, not first_round)
|
|
first_round = False
|
|
if len(active) <= 3:
|
|
# Quoted data most often leaves three survivors: the
|
|
# true dialect, its equally consistent unquoted shadow,
|
|
# and one accident. Parsing the whole sample with them
|
|
# is cheaper than another elimination round.
|
|
window = len(sample)
|
|
else:
|
|
# Too small a factor would increase the total
|
|
# re-parsing cost, too large -- the cost of the next
|
|
# round if this one did not eliminate enough.
|
|
window *= 4
|
|
|
|
best = None
|
|
best_score = None
|
|
for combo, rows in sorted(parsed):
|
|
if rows is None:
|
|
continue
|
|
_, delimiter, quotechar, escapechar = combo
|
|
nfields, share = self._modal_share(rows)
|
|
if nfields < 2:
|
|
# The delimiter does not delimit anything.
|
|
continue
|
|
try:
|
|
preference = -self.preferred.index(delimiter)
|
|
except ValueError:
|
|
preference = -len(self.preferred)
|
|
# A successful quoted parse is direct evidence; the preferred
|
|
# delimiters list is only a nudge.
|
|
score = (share, len(rows), bool(quotechar), preference)
|
|
if best_score is None or score > best_score:
|
|
best_score = score
|
|
best = combo[1:]
|
|
|
|
if best is None:
|
|
raise Error("Could not determine delimiter")
|
|
delimiter, quotechar, escapechar = best
|
|
doublequote = self._detect_doublequote(lines, *best)
|
|
skipinitialspace = self._detect_skipinitialspace(lines, *best,
|
|
doublequote)
|
|
|
|
class dialect(Dialect):
|
|
_name = "sniffed"
|
|
lineterminator = '\r\n'
|
|
quoting = QUOTE_MINIMAL
|
|
|
|
dialect.delimiter = delimiter
|
|
# _csv.reader won't accept a quotechar of ''
|
|
dialect.quotechar = quotechar or '"'
|
|
dialect.escapechar = escapechar or None
|
|
dialect.doublequote = doublequote
|
|
dialect.skipinitialspace = skipinitialspace
|
|
|
|
return dialect
|
|
|
|
def _parses_as_single_column(self, sample):
|
|
"""
|
|
True if the whole sample parses as a single column of quoted
|
|
fields (the last one may be cut off in the middle), so there
|
|
is no delimiter to find.
|
|
"""
|
|
import re
|
|
|
|
for q in '"', "'":
|
|
if q in sample:
|
|
row_re = (fr' *+{q}(?:[^{q}]|{q}{q})*+'
|
|
fr'(?:{q} *+(?:[\r\n]++|\z)|\z)')
|
|
if re.fullmatch(fr'(?:{row_re})++', sample):
|
|
return True
|
|
return False
|
|
|
|
def _split_dormant(self, part, quotechar, combos):
|
|
"""
|
|
Split the dormant combinations into those ready for trial
|
|
parsing and the rest.
|
|
|
|
A combination is ready when its quote character occurs in
|
|
*part* at the start of a field, i.e. at the start of a line or
|
|
after its delimiter; until then parsing would not differ from
|
|
the unquoted variant. Spaces before the quote are allowed even
|
|
for the space delimiter: a false activation only costs a trial
|
|
parse.
|
|
"""
|
|
import re
|
|
|
|
remaining = {combo[1] for combo in combos}
|
|
found = set()
|
|
pos = 0
|
|
while remaining:
|
|
# Include only the delimiters not found yet, so that the
|
|
# search skips over the found ones; the compiled patterns
|
|
# come from the re cache.
|
|
cls = re.escape(''.join(sorted(remaining)))
|
|
m = re.compile(fr'(?:^|([\r\n{cls}]))'
|
|
fr' *{quotechar}').search(part, pos)
|
|
if m is None:
|
|
break
|
|
pre = m[1]
|
|
if pre is None or pre in '\r\n':
|
|
# A quote at the start of a line starts a field for
|
|
# every delimiter.
|
|
found |= remaining
|
|
break
|
|
found.add(pre)
|
|
remaining.discard(pre)
|
|
pos = m.end()
|
|
activated = [combo for combo in combos if combo[1] in found]
|
|
still_dormant = [combo for combo in combos if combo[1] not in found]
|
|
return activated, still_dormant
|
|
|
|
def _make_reader(self, lines, delimiter, quotechar, escapechar,
|
|
doublequote=True, skipinitialspace=None):
|
|
"""
|
|
Create a reader for trial parsing. quotechar '' means no
|
|
quoting and escapechar '' means no escape character.
|
|
"""
|
|
if skipinitialspace is None:
|
|
# Be lenient to spaces after a delimiter, unless the
|
|
# delimiter is a space itself.
|
|
skipinitialspace = delimiter != ' '
|
|
return reader(lines, delimiter=delimiter,
|
|
quotechar=quotechar or '"',
|
|
quoting=QUOTE_MINIMAL if quotechar else QUOTE_NONE,
|
|
escapechar=escapechar or None,
|
|
doublequote=doublequote,
|
|
skipinitialspace=skipinitialspace,
|
|
strict=True)
|
|
|
|
def _try_dialect(self, lines, cut, delimiter, quotechar, escapechar):
|
|
"""
|
|
Parse the sample, pre-split into *lines*, and return the list
|
|
of the number of fields in every parsed row, or None if not a
|
|
single row was parsed.
|
|
|
|
If the sample cannot be parsed to the end (for example it is
|
|
cut off in the middle of a quoted field, or the combination
|
|
does not fit the sample), the rows parsed so far are counted.
|
|
The last row is not counted if *cut* is true: the sample can
|
|
be cut off in the middle of it.
|
|
"""
|
|
rows = []
|
|
try:
|
|
rows.extend(map(len, self._make_reader(lines, delimiter,
|
|
quotechar, escapechar)))
|
|
except Error:
|
|
# The row which failed to parse is not counted.
|
|
pass
|
|
else:
|
|
if cut and len(rows) > 1:
|
|
rows.pop()
|
|
if 0 in rows:
|
|
# Blank lines produce empty rows.
|
|
rows = [nfields for nfields in rows if nfields]
|
|
return rows or None
|
|
|
|
def _eliminate_worse(self, parsed, judge_hopeless):
|
|
"""
|
|
Return the combinations from *parsed* (a list of (combination,
|
|
rows) pairs) without those which are clearly worse than the
|
|
leader. Combinations with too few parsed rows (e.g. if the
|
|
parsed part ends in the middle of a large quoted field) are
|
|
not judged yet.
|
|
|
|
If *judge_hopeless* is false, keep the combinations whose
|
|
delimiter does not delimit anything. Unlike the comparison
|
|
with the leader, which self-normalizes when the parsed part is
|
|
not representative, this verdict is absolute and irreversible,
|
|
so it is not trusted to the first part, which covers the least
|
|
representative beginning of the sample (titles, headers,
|
|
preamble).
|
|
"""
|
|
# Judging a combination by fewer rows is too noisy.
|
|
min_rows = 16
|
|
hopeless = set()
|
|
scores = {}
|
|
for combo, rows in parsed:
|
|
if rows is not None and len(rows) >= min_rows:
|
|
nfields, share = self._modal_share(rows)
|
|
if nfields < 2:
|
|
if judge_hopeless:
|
|
hopeless.add(combo)
|
|
else:
|
|
scores[combo] = share
|
|
threshold = max(scores.values(), default=0.0) - 0.1
|
|
return [combo for combo, _ in parsed
|
|
if combo not in hopeless
|
|
and scores.get(combo, threshold) >= threshold]
|
|
|
|
def _modal_share(self, rows):
|
|
"""
|
|
The most common number of fields in a row and its share of all
|
|
rows. Prefer the smaller number of fields in the case of a
|
|
tie: a candidate delimiter which delimits only half of the rows
|
|
is not convincing.
|
|
"""
|
|
from collections import Counter
|
|
|
|
counts = Counter(rows)
|
|
nfields = max(counts, key=lambda n: (counts[n], -n))
|
|
return nfields, counts[nfields] / len(rows)
|
|
|
|
def _detect_doublequote(self, lines, delimiter, quotechar, escapechar):
|
|
"""
|
|
True if a doubled quote character represents a single quote
|
|
character in the sample: interpreting it so changes the result
|
|
of parsing.
|
|
"""
|
|
if not quotechar or not any(quotechar * 2 in line
|
|
for line in lines):
|
|
return False
|
|
readers = [self._make_reader(
|
|
lines, delimiter, quotechar, escapechar,
|
|
doublequote=doublequote)
|
|
for doublequote in (False, True)]
|
|
while True:
|
|
rows = []
|
|
for rdr in readers:
|
|
try:
|
|
rows.append(next(rdr))
|
|
except (StopIteration, Error):
|
|
# Ending cleanly and failing are equivalent here:
|
|
# after equal rows both readers are at the same
|
|
# position, so they cannot end for different
|
|
# reasons.
|
|
rows.append(None)
|
|
if rows[0] != rows[1]:
|
|
return True
|
|
if rows == [None, None]:
|
|
return False
|
|
|
|
def _detect_skipinitialspace(self, lines, delimiter, quotechar,
|
|
escapechar, doublequote):
|
|
"""
|
|
Detect whether the spaces following a delimiter are a part of
|
|
the format or of the data.
|
|
"""
|
|
results = []
|
|
for skipinitialspace in False, True:
|
|
rows = []
|
|
try:
|
|
rows.extend(self._make_reader(
|
|
lines, delimiter, quotechar, escapechar,
|
|
doublequote=doublequote,
|
|
skipinitialspace=skipinitialspace))
|
|
except Error:
|
|
# Keep the rows parsed before the error.
|
|
pass
|
|
results.append([row for row in rows if row])
|
|
if results[0] == results[1]:
|
|
return False # No evidence.
|
|
counts = [[len(row) for row in rows] for rows in results]
|
|
if counts[0] != counts[1]:
|
|
# Prefer the more consistent row widths.
|
|
return len(set(counts[1])) <= len(set(counts[0]))
|
|
# Only some spaces are stripped. A field differs only if
|
|
# a space was skipped at its start, which tells the padding
|
|
# apart from the spaces inside quoted or escaped fields.
|
|
if not all(kept_field != skipped_field
|
|
for kept_row, skipped_row in zip(*results)
|
|
for kept_field, skipped_field in zip(kept_row[1:],
|
|
skipped_row[1:])):
|
|
return False
|
|
# The first field of a row is commonly not padded ('a, b, c'),
|
|
# so the first fields need only agree with each other.
|
|
first = [kept_row[0] != skipped_row[0]
|
|
for kept_row, skipped_row in zip(*results)]
|
|
return all(first) or not any(first)
|
|
|
|
def has_header(self, sample):
|
|
# Creates a dictionary of types of data in each column. If any
|
|
# column is of a single type (say, integers), *except* for the first
|
|
# row, then the first row is presumed to be labels. If the type
|
|
# can't be determined, it is assumed to be a string in which case
|
|
# the length of the string is the determining factor: if all of the
|
|
# rows except for the first are the same length, it's a header.
|
|
# Finally, a 'vote' is taken at the end for each column, adding or
|
|
# subtracting from the likelihood of the first row being a header.
|
|
|
|
rdr = reader(StringIO(sample), self.sniff(sample))
|
|
|
|
header = next(rdr) # assume first row is header
|
|
|
|
columns = len(header)
|
|
columnTypes = {}
|
|
for i in range(columns): columnTypes[i] = None
|
|
|
|
checked = 0
|
|
for row in rdr:
|
|
# arbitrary number of rows to check, to keep it sane
|
|
if checked > 20:
|
|
break
|
|
checked += 1
|
|
|
|
if len(row) != columns:
|
|
continue # skip rows that have irregular number of columns
|
|
|
|
for col in list(columnTypes.keys()):
|
|
thisType = complex
|
|
try:
|
|
thisType(row[col])
|
|
except (ValueError, OverflowError):
|
|
# fallback to length of string
|
|
thisType = len(row[col])
|
|
|
|
if thisType != columnTypes[col]:
|
|
if columnTypes[col] is None: # add new column type
|
|
columnTypes[col] = thisType
|
|
else:
|
|
# type is inconsistent, remove column from
|
|
# consideration
|
|
del columnTypes[col]
|
|
|
|
# finally, compare results against first row and "vote"
|
|
# on whether it's a header
|
|
hasHeader = 0
|
|
for col, colType in columnTypes.items():
|
|
if isinstance(colType, int): # it's a length
|
|
if len(header[col]) != colType:
|
|
hasHeader += 1
|
|
else:
|
|
hasHeader -= 1
|
|
else: # attempt typecast
|
|
try:
|
|
colType(header[col])
|
|
except (ValueError, TypeError):
|
|
hasHeader += 1
|
|
else:
|
|
hasHeader -= 1
|
|
|
|
return hasHeader > 0
|
|
|
|
|
|
def __getattr__(name):
|
|
if name == "__version__":
|
|
from warnings import _deprecated
|
|
|
|
_deprecated("__version__", remove=(3, 20))
|
|
return "1.0" # Do not change
|
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|