feat: Added BOM capability for output files (#1267)

csvkit/cli.py

@@ -19,6 +19,7 @@
 from agate.data_types.base import DEFAULT_NULL_VALUES
 
 from csvkit.exceptions import ColumnIdentifierError, RequiredHeaderError
+from csvkit.features.AddBom import AddBOM
 
 try:
     import zstandard
@@ -134,6 +135,8 @@ def run(self):
         if 'f' not in self.override_flags:
             self.input_file = self._open_input_file(self.args.input_path)
 
+        AddBOM.run(self.output_file, self.args)
+
         try:
             with warnings.catch_warnings():
                 if getattr(self.args, 'no_header_row', None):
@@ -245,6 +248,8 @@ def _init_common_parser(self):
                 help='Insert a column of line numbers at the front of the output. Useful when piping to grep or as a '
                      'simple primary key.')
 
+        AddBOM.argument(self.argparser,self)
+
         # Input/Output
         if 'zero' not in self.override_flags:
             self.argparser.add_argument(

csvkit/features/AddBom.py

@@ -0,0 +1,54 @@
+from argparse import ArgumentParser, Namespace
+from io import TextIOWrapper, BytesIO
+from typing import Union
+
+
+class AddBOM:
+
+    @staticmethod
+    def _get_BOM() -> bytes:
+
+        from codecs import BOM_UTF8
+
+        return BOM_UTF8
+
+    @staticmethod
+    def enabled(arguments: Union[Namespace,list, None] = None) -> bool:
+
+        if isinstance(arguments, Namespace) or isinstance(arguments, list):
+            return "add_bom" in arguments and arguments.add_bom
+
+        return False
+
+    @staticmethod
+    def argument(arguments: ArgumentParser, utility: object):
+
+        # These string usage to validate the class  is an architecture
+        # fail as is not possible to check the class type before
+        # is initialized
+
+        if "SQL2CSV" in str(utility.__class__):
+            return
+
+        if "CSVPy" in str(utility.__class__):
+            return
+
+        arguments.add_argument(
+            "--add-bom",
+            dest="add_bom",
+            action="store_true",
+            default=False,
+            help="Add Byte Order Mark (BOM) to the output",
+        )
+
+    @staticmethod
+    def run(
+        output: TextIOWrapper,
+        arguments: Union[Namespace,list, None] = None,
+    ):
+
+        if not AddBOM.enabled(arguments):
+            return
+
+        BOM = AddBOM._get_BOM()
+        output.buffer.write(BOM)