| |
| 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"] |
| |
| |
| _dialect_attributes = frozenset({ |
| 'delimiter', 'quotechar', 'escapechar', 'doublequote', |
| 'skipinitialspace', 'lineterminator', 'quoting', 'strict', |
| }) |
| |
| 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 |
| |
| def __replace__(self, /, **changes): |
| unexpected = changes.keys() - _dialect_attributes |
| if unexpected: |
| raise TypeError(f'__replace__() got an unexpected keyword ' |
| f'argument {min(unexpected)!r}') |
| new = object.__new__(self.__class__) |
| new.__dict__.update(self.__dict__) |
| new.__dict__.update(changes) |
| new._validate() |
| return new |
| |
| 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" |
| quoting = QUOTE_MINIMAL |
| |
| dialect.lineterminator = self._detect_lineterminator(lines) |
| 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 _detect_lineterminator(self, lines): |
| """ |
| Detect the line terminator by majority vote among the line |
| endings. A line break inside a quoted field is counted too, |
| but it takes more of them than of the real ones to win the |
| vote. A tie is broken in the order '\\r\\n', '\\n', '\\r', |
| so a sample without a complete line gives '\\r\\n'. |
| """ |
| counts = dict.fromkeys(('\r\n', '\n', '\r'), 0) |
| for line in lines: |
| for lineterminator in counts: |
| if line.endswith(lineterminator): |
| counts[lineterminator] += 1 |
| break |
| # max() returns the first of equal candidates, and dict |
| # preserves the insertion order. |
| return max(counts, key=counts.get) |
| |
| 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}") |
