1# -*- coding: utf-8 -*- 2 3# Copyright: (c) 2021, Andrew Pantuso (@ajpantuso) <ajpantuso@gmail.com> 4# Copyright: (c) 2018, Dag Wieers (@dagwieers) <dag@wieers.com> 5# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 6 7from __future__ import absolute_import, division, print_function 8__metaclass__ = type 9 10import csv 11from io import BytesIO, StringIO 12 13from ansible.module_utils.common.text.converters import to_native 14from ansible.module_utils.six import PY3 15 16 17class CustomDialectFailureError(Exception): 18 pass 19 20 21class DialectNotAvailableError(Exception): 22 pass 23 24 25CSVError = csv.Error 26 27 28def initialize_dialect(dialect, **kwargs): 29 # Add Unix dialect from Python 3 30 class unix_dialect(csv.Dialect): 31 """Describe the usual properties of Unix-generated CSV files.""" 32 delimiter = ',' 33 quotechar = '"' 34 doublequote = True 35 skipinitialspace = False 36 lineterminator = '\n' 37 quoting = csv.QUOTE_ALL 38 39 csv.register_dialect("unix", unix_dialect) 40 41 if dialect not in csv.list_dialects(): 42 raise DialectNotAvailableError("Dialect '%s' is not supported by your version of python." % dialect) 43 44 # Create a dictionary from only set options 45 dialect_params = dict((k, v) for k, v in kwargs.items() if v is not None) 46 if dialect_params: 47 try: 48 csv.register_dialect('custom', dialect, **dialect_params) 49 except TypeError as e: 50 raise CustomDialectFailureError("Unable to create custom dialect: %s" % to_native(e)) 51 dialect = 'custom' 52 53 return dialect 54 55 56def read_csv(data, dialect, fieldnames=None): 57 58 data = to_native(data, errors='surrogate_or_strict') 59 60 if PY3: 61 fake_fh = StringIO(data) 62 else: 63 fake_fh = BytesIO(data) 64 65 reader = csv.DictReader(fake_fh, fieldnames=fieldnames, dialect=dialect) 66 67 return reader 68