6507626273

Committed 13 Oct 2023 11:25AM UTC coverage: 42.006% (-53.7%) from 95.72%

Build # 6507626273

Build Type

Pull #433

github

Committed by

odscjames

Commit Message

New "Geo" optional dependencies

https://github.com/OpenDataServices/flatten-tool/issues/424

Pull Request Pull Request #433: New "Geo" optional dependencies

Run Details

38 of 38 new or added lines in 6 files covered. (100.0%)

1466 of 3490 relevant lines covered (42.01%)

4.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.0

/flattentool/output.py

"""
Code to output a parsed flattened JSON schema in various spreadsheet
formats.

"""

import csv
import os
from warnings import warn

import odf.table
import odf.text
import openpyxl
from odf.opendocument import OpenDocumentSpreadsheet
from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE

from flattentool.exceptions import DataErrorWarning
from flattentool.i18n import _


class SpreadsheetOutput(object):
    # output_name is given a default here, partly to help with tests,
    # but should have been defined by the time we get here.
    def __init__(
        self,
        parser,
        main_sheet_name="main",
        output_name="unflattened",
        sheet_prefix="",
        line_terminator="\r\n",
    ):
        self.parser = parser
        self.main_sheet_name = main_sheet_name
        self.output_name = output_name
        self.sheet_prefix = sheet_prefix
        self.line_terminator = line_terminator

    def open(self):
        pass

    def write_sheet(self, sheet_name, sheet_header, sheet_lines=None):
        raise NotImplementedError

    def write_sheets(self):
        self.open()

        self.write_sheet(self.main_sheet_name, self.parser.main_sheet)
        for sheet_name, sub_sheet in sorted(self.parser.sub_sheets.items()):
            self.write_sheet(sheet_name, sub_sheet)

        self.close()

    def close(self):
        pass


class XLSXOutput(SpreadsheetOutput):
    def open(self):
        # write only means that the output will be streamed
        self.workbook = openpyxl.Workbook(write_only=True)

    def write_sheet(self, sheet_name, sheet):
        sheet_header = list(sheet)
        worksheet = self.workbook.create_sheet()
        worksheet.title = (self.sheet_prefix + sheet_name)[:31]
        worksheet.append(sheet_header)
        for sheet_line in sheet.lines:
            line = []
            for header in sheet_header:
                value = sheet_line.get(header)
                if isinstance(value, str):
                    new_value = ILLEGAL_CHARACTERS_RE.sub("", value)
                    if new_value != value:
                        warn(
                            _(
                                "Character(s) in '{}' are not allowed in a spreadsheet cell. Those character(s) will be removed"
                            ).format(value),
                            DataErrorWarning,
                        )
                    value = new_value
                line.append(value)
            worksheet.append(line)

    def close(self):
        self.workbook.save(self.output_name)


class CSVOutput(SpreadsheetOutput):
    def open(self):
        try:
            os.makedirs(self.output_name)
        except OSError:
            pass

    def write_sheet(self, sheet_name, sheet):
        sheet_header = list(sheet)
        with open(
            os.path.join(self.output_name, self.sheet_prefix + sheet_name + ".csv"),
            "w",
            newline="",
            encoding="utf-8",
        ) as csv_file:
            dictwriter = csv.DictWriter(
                csv_file, sheet_header, lineterminator=self.line_terminator
            )
            dictwriter.writeheader()
            for sheet_line in sheet.lines:
                dictwriter.writerow(sheet_line)


class ODSOutput(SpreadsheetOutput):
    def open(self):
        self.workbook = OpenDocumentSpreadsheet()

    def _make_cell(self, value):
        """Util for creating an ods cell"""

        if value:
            try:
                # See if value parses as a float
                cell = odf.table.TableCell(valuetype="float", value=float(value))
            except ValueError:
                cell = odf.table.TableCell(valuetype="string")
        else:
            cell = odf.table.TableCell(valuetype="Nonetype")

        p = odf.text.P(text=value)
        cell.addElement(p)

        return cell

    def write_sheet(self, sheet_name, sheet):

        worksheet = odf.table.Table(name=(self.sheet_prefix + sheet_name)[:31])
        sheet_header = list(sheet)

        header_row = odf.table.TableRow()

        for header in sheet_header:
            header_row.addElement(self._make_cell(header))

        worksheet.addElement(header_row)

        for sheet_line in sheet.lines:
            row = odf.table.TableRow()
            for header in sheet_header:
                value = sheet_line.get(header)
                if isinstance(value, str):
                    new_value = ILLEGAL_CHARACTERS_RE.sub("", value)
                    if new_value != value:
                        warn(
                            _(
                                "Character(s) in '{}' are not allowed in a spreadsheet cell. Those character(s) will be removed"
                            ).format(value),
                            DataErrorWarning,
                        )
                    value = new_value
                row.addElement(self._make_cell(value))
            worksheet.addElement(row)

        self.workbook.spreadsheet.addElement(worksheet)

    def close(self):
        self.workbook.save(self.output_name)


FORMATS = {"xlsx": XLSXOutput, "csv": CSVOutput, "ods": ODSOutput}

FORMATS_SUFFIX = {
    "xlsx": ".xlsx",
    "ods": ".ods",
    "csv": "",  # This is the suffix for the directory
}

LINE_TERMINATORS = {"LF": "\n", "CRLF": "\r\n"}

1	"""
2	Code to output a parsed flattened JSON schema in various spreadsheet
3	formats.
4
5	"""
6
7	import csv	10✔
8	import os	10✔
9	from warnings import warn	10✔
10
11	import odf.table	10✔
12	import odf.text	10✔
13	import openpyxl	10✔
14	from odf.opendocument import OpenDocumentSpreadsheet	10✔
15	from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE	10✔
16
17	from flattentool.exceptions import DataErrorWarning	10✔
18	from flattentool.i18n import _	10✔
19
20
21	class SpreadsheetOutput(object):	10✔
22	# output_name is given a default here, partly to help with tests,
23	# but should have been defined by the time we get here.
24	def __init__(	10✔
25	self,
26	parser,
27	main_sheet_name="main",
28	output_name="unflattened",
29	sheet_prefix="",
30	line_terminator="\r\n",
31	):
32	self.parser = parser	10✔
33	self.main_sheet_name = main_sheet_name	10✔
34	self.output_name = output_name	10✔
35	self.sheet_prefix = sheet_prefix	10✔
36	self.line_terminator = line_terminator	10✔
37
38	def open(self):	10✔
39	pass	×
40
41	def write_sheet(self, sheet_name, sheet_header, sheet_lines=None):	10✔
42	raise NotImplementedError	×
43
44	def write_sheets(self):	10✔
45	self.open()	10✔
46
47	self.write_sheet(self.main_sheet_name, self.parser.main_sheet)	10✔
48	for sheet_name, sub_sheet in sorted(self.parser.sub_sheets.items()):	10✔
49	self.write_sheet(sheet_name, sub_sheet)	×
50
51	self.close()	10✔
52
53	def close(self):	10✔
54	pass	10✔
55
56
57	class XLSXOutput(SpreadsheetOutput):	10✔
58	def open(self):	10✔
59	# write only means that the output will be streamed
60	self.workbook = openpyxl.Workbook(write_only=True)	10✔
61
62	def write_sheet(self, sheet_name, sheet):	10✔
63	sheet_header = list(sheet)	10✔
64	worksheet = self.workbook.create_sheet()	10✔
65	worksheet.title = (self.sheet_prefix + sheet_name)[:31]	10✔
66	worksheet.append(sheet_header)	10✔
67	for sheet_line in sheet.lines:	10✔
68	line = []	×
69	for header in sheet_header:	×
70	value = sheet_line.get(header)	×
71	if isinstance(value, str):	×
72	new_value = ILLEGAL_CHARACTERS_RE.sub("", value)	×
73	if new_value != value:	×
74	warn(	×
75	_(
76	"Character(s) in '{}' are not allowed in a spreadsheet cell. Those character(s) will be removed"
77	).format(value),
78	DataErrorWarning,
79	)
80	value = new_value	×
81	line.append(value)	×
82	worksheet.append(line)	×
83
84	def close(self):	10✔
85	self.workbook.save(self.output_name)	10✔
86
87
88	class CSVOutput(SpreadsheetOutput):	10✔
89	def open(self):	10✔
90	try:	10✔
91	os.makedirs(self.output_name)	10✔
92	except OSError:	10✔
93	pass	10✔
94
95	def write_sheet(self, sheet_name, sheet):	10✔
96	sheet_header = list(sheet)	10✔
97	with open(	10✔
98	os.path.join(self.output_name, self.sheet_prefix + sheet_name + ".csv"),
99	"w",
100	newline="",
101	encoding="utf-8",
102	) as csv_file:
103	dictwriter = csv.DictWriter(	10✔
104	csv_file, sheet_header, lineterminator=self.line_terminator
105	)
106	dictwriter.writeheader()	10✔
107	for sheet_line in sheet.lines:	10✔
108	dictwriter.writerow(sheet_line)	10✔
109
110
111	class ODSOutput(SpreadsheetOutput):	10✔
112	def open(self):	10✔
113	self.workbook = OpenDocumentSpreadsheet()	10✔
114
115	def _make_cell(self, value):	10✔
116	"""Util for creating an ods cell"""
117
118	if value:	10✔
119	try:	10✔
120	# See if value parses as a float
121	cell = odf.table.TableCell(valuetype="float", value=float(value))	10✔
122	except ValueError:	10✔
123	cell = odf.table.TableCell(valuetype="string")	10✔
124	else:
125	cell = odf.table.TableCell(valuetype="Nonetype")	×
126
127	p = odf.text.P(text=value)	10✔
128	cell.addElement(p)	10✔
129
130	return cell	10✔
131
132	def write_sheet(self, sheet_name, sheet):	10✔
133
134	worksheet = odf.table.Table(name=(self.sheet_prefix + sheet_name)[:31])	10✔
135	sheet_header = list(sheet)	10✔
136
137	header_row = odf.table.TableRow()	10✔
138
139	for header in sheet_header:	10✔
140	header_row.addElement(self._make_cell(header))	10✔
141
142	worksheet.addElement(header_row)	10✔
143
144	for sheet_line in sheet.lines:	10✔
145	row = odf.table.TableRow()	×
146	for header in sheet_header:	×
147	value = sheet_line.get(header)	×
148	if isinstance(value, str):	×
149	new_value = ILLEGAL_CHARACTERS_RE.sub("", value)	×
150	if new_value != value:	×
151	warn(	×
152	_(
153	"Character(s) in '{}' are not allowed in a spreadsheet cell. Those character(s) will be removed"
154	).format(value),
155	DataErrorWarning,
156	)
157	value = new_value	×
158	row.addElement(self._make_cell(value))	×
159	worksheet.addElement(row)	×
160
161	self.workbook.spreadsheet.addElement(worksheet)	10✔
162
163	def close(self):	10✔
164	self.workbook.save(self.output_name)	10✔
165
166
167	FORMATS = {"xlsx": XLSXOutput, "csv": CSVOutput, "ods": ODSOutput}	10✔
168
169	FORMATS_SUFFIX = {	10✔
170	"xlsx": ".xlsx",
171	"ods": ".ods",
172	"csv": "", # This is the suffix for the directory
173	}
174
175	LINE_TERMINATORS = {"LF": "\n", "CRLF": "\r\n"}	10✔

OpenDataServices / flatten-tool / 6507626273

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous