95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from __future__ import annotations
|
|
from typing import Iterator, Dict, Any, Tuple, TextIO, List
|
|
import re
|
|
import datetime
|
|
import csv
|
|
|
|
RE_INT = re.compile(r'-?[0-9]+')
|
|
RE_FLOAT = re.compile(r'-?[0-9]+\.[0-9]+')
|
|
RE_STR_START = re.compile(r'.*,"[^"]*$')
|
|
RE_STR_END = re.compile(r'^[^"]*",.*')
|
|
|
|
|
|
def cast_value(value: str) -> Any:
|
|
if value == '':
|
|
return None
|
|
elif value[0] == '"' and value[-1] == '"':
|
|
return value[1:-1].replace('""', '"')
|
|
elif value == 'T':
|
|
return True
|
|
elif value == 'F':
|
|
return False
|
|
elif RE_INT.fullmatch(value):
|
|
return int(value)
|
|
elif RE_FLOAT.fullmatch(value):
|
|
return float(value)
|
|
elif len(value) == 10 and value[4] == '-' and value[7] == '-':
|
|
return datetime.datetime.strptime(value, '%Y-%m-%d').date()
|
|
elif len(value) == 8 and value[2] == ':' and value[5] == ':':
|
|
return datetime.time.fromisoformat(value)
|
|
else:
|
|
raise RuntimeError(f'unable to infer type of value "{value}"')
|
|
|
|
|
|
def convert_value(value: Any, table: str = None, column: str = None) -> str:
|
|
if value is None:
|
|
return ''
|
|
if type(value) == str:
|
|
return '"' + value.replace('"', '""') + '"'
|
|
elif type(value) == bool:
|
|
return 'T' if value else 'F'
|
|
elif type(value) == datetime.datetime and table is not None and column is not None:
|
|
if value.year == 1899 and value.month == 12 and value.day == 30:
|
|
return value.strftime('%H:%M:%S')
|
|
elif value.hour == 0 and value.minute == 0 and value.second == 0:
|
|
return value.strftime('%Y-%m-%d')
|
|
return str(value)
|
|
|
|
|
|
class CsvFile:
|
|
file: TextIO
|
|
reader: csv.reader
|
|
writer: csv.writer
|
|
|
|
def __init__(self, file: TextIO):
|
|
self.file = file
|
|
self.writer = csv.writer(self.file, doublequote=False, quoting=csv.QUOTE_NONE, escapechar='\\', quotechar=None)
|
|
self.reader = csv.reader(self.file, doublequote=False, quoting=csv.QUOTE_NONE, escapechar='\\', quotechar=None)
|
|
|
|
def __enter__(self) -> CsvFile:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
|
|
return self.file.__exit__(exc_type, exc_val, exc_tb)
|
|
|
|
def __iter__(self) -> Iterator[List[str]]:
|
|
return self.reader.__iter__()
|
|
|
|
def row(self, *fields, raw: bool = False):
|
|
if not raw:
|
|
fields = (convert_value(field) for field in fields)
|
|
self.writer.writerow(fields)
|
|
|
|
def header(self, *headers):
|
|
self.writer.writerow(str(h) for h in headers)
|
|
|
|
|
|
def csv_open(filename: str, mode: str = 'w+') -> CsvFile:
|
|
return CsvFile(open(filename, mode, encoding='utf-8', newline=''))
|
|
|
|
|
|
def csv_parse(filename: str) -> Iterator[Tuple]:
|
|
with csv_open(filename, 'r') as f:
|
|
rows = f.__iter__()
|
|
yield tuple([field.strip() for field in next(rows)])
|
|
yield from (tuple(cast_value(field) for field in row) for row in rows)
|
|
|
|
|
|
def csv_parse_dict(filename: str) -> Iterator[Dict[str, Any]]:
|
|
rows = csv_parse(filename)
|
|
header = next(rows)
|
|
return ({header[i]: part for i, part in enumerate(row)} for row in rows)
|