-
Notifications
You must be signed in to change notification settings - Fork 259
/
table.py
55 lines (49 loc) · 2.23 KB
/
table.py
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
import logging
import csv
from pathlib import Path
from followthemoney.types import registry
log = logging.getLogger(__name__)
class TableSupport(object):
"""Handle creating rows from an ingestor."""
def set_column_headers(self, table, headers):
table.set('columns', [headers])
def emit_row_dicts(self, table, rows):
csv_path = Path(self.manager.work_path).joinpath(table.id + ".csv")
with open(csv_path, 'w', encoding='utf-8') as fp:
csv_writer = csv.writer(fp)
headers = []
for index, row in enumerate(rows, 1):
if not headers:
headers = list(row.keys())
values = list(row.values())
csv_writer.writerow(values)
entity = self.manager.make_entity('Row')
entity.make_id(table.id, index)
entity.set('index', index)
entity.set('cells', registry.json.pack(values))
entity.set('table', table)
self.manager.emit_entity(entity)
self.manager.emit_text_fragment(table, values, entity.id)
self.set_column_headers(table, headers)
csv_hash = self.manager.archive_store(csv_path)
table.set("csvHash", csv_hash)
def emit_row_tuples(self, table, rows):
csv_path = Path(self.manager.work_path).joinpath(table.id + ".csv")
with open(csv_path, 'w', encoding='utf-8') as fp:
csv_writer = csv.writer(fp)
headers = []
for index, row in enumerate(rows, 1):
if not headers:
headers = ["Column %s" % i for i in range(1, len(row) + 1)]
row = list(row)
csv_writer.writerow(row)
entity = self.manager.make_entity('Row')
entity.make_id(table.id, index)
entity.set('index', index)
entity.add('cells', registry.json.pack(row))
entity.add('table', table)
self.manager.emit_entity(entity)
self.manager.emit_text_fragment(table, row, entity.id)
self.set_column_headers(table, headers)
csv_hash = self.manager.archive_store(csv_path)
table.set("csvHash", csv_hash)