promethium-python-compatibility/csv.py at master · remobjects/promethium-python-compatibility · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@namespace("csv")

from Promethium import List
from collections import OrderedDict

# A small, opt-in subset of Python's csv module: `parse_line`/`write_row`
# for one row at a time — not a `reader`/`writer` object bound to a file
# handle (this project's own scope statement excludes filesystem APIs;
# see the README's "Build notes"). Handles the common cases: comma-
# separated fields, double-quoted fields (needed when a field itself
# contains a comma), and `""` as an escaped quote inside a quoted field —
# CPython's default `excel` dialect's escaping rule. Does not support a
# custom delimiter/quote character, embedded newlines inside a quoted
# field, or any of the other `Dialect` options.
#
# Built on the same manual-character-scan technique `string.py`'s
# `capwords` and `textwrap.py`'s `_words` already use (no `.Split`, no
# regex — see `string.py`'s notes on why).


def parse_line(line: str) -> List[str]:
    fields: List[str] = List[str]()
    length: int = _strutil.length(line)
    index: int = 0
    field: str = ""
    inQuotes: bool = False
    while index < length:
        ch: str = _strutil.substring(line, index, 1)
        if inQuotes:
            if ch == "\"":
                if index + 1 < length and _strutil.substring(line, index + 1, 1) == "\"":
                    field += "\""
                    index += 1
                else:
                    inQuotes = False
            else:
                field += ch
        else:
            if ch == "\"":
                inQuotes = True
            elif ch == ",":
                fields.append(field)
                field = ""
            else:
                field += ch
        index += 1
    fields.append(field)
    return fields


def _fieldNeedsQuoting(field: str) -> bool:
    length: int = _strutil.length(field)
    index: int = 0
    while index < length:
        ch: str = _strutil.substring(field, index, 1)
        if ch == "," or ch == "\"" or ch == "\n" or ch == "\r":
            return True
        index += 1
    return False


def _quoteField(field: str) -> str:
    result: str = ""
    length: int = _strutil.length(field)
    index: int = 0
    while index < length:
        ch: str = _strutil.substring(field, index, 1)
        if ch == "\"":
            result += "\"\""
        else:
            result += ch
        index += 1
    return "\"" + result + "\""


def write_row(fields: List[str]) -> str:
    result: str = ""
    index: int = 0
    while index < len(fields):
        if index > 0:
            result += ","
        field: str = fields.__getitem__(index)
        if _fieldNeedsQuoting(field):
            result += _quoteField(field)
        else:
            result += field
        index += 1
    return result


# `DictReader`/`DictWriter`-equivalent, one row at a time (same row-at-a-
# time scope as `parse_line`/`write_row` above, not a file-bound object).
# Missing trailing fields read as `""` (matching CPython's own `restval`
# default of `None`, adapted to a concrete `str` the way this project
# elsewhere stands in for CPython's `None` defaults); a header key with
# no matching value in the row to write is skipped and written as `""`.


def parse_line_dict(header: List[str], line: str) -> OrderedDict[str, str]:
    values: List[str] = parse_line(line)
    result: OrderedDict[str, str] = OrderedDict[str, str]()
    valueCount: int = values.__len__()
    i: int = 0
    n: int = header.__len__()
    while i < n:
        if i < valueCount:
            result[header[i]] = values[i]
        else:
            result[header[i]] = ""
        i += 1
    return result


def write_row_dict(header: List[str], row: OrderedDict[str, str]) -> str:
    values: List[str] = List[str]()
    i: int = 0
    n: int = header.__len__()
    while i < n:
        key: str = header[i]
        if key in row:
            values.append(row[key])
        else:
            values.append("")
        i += 1
    return write_row(values)