Escaping ASNI escape sequences in stdout (#17)

* escaping windows terminal sequence * trigger workflow Authored-by: Jan Nitschke <[email protected]>
Dashlane · Feb 21, 2022 · 77cce0d · 77cce0d
1 parent 949c0eb
commit 77cce0d
Showing 1 changed file with 13 additions and 2 deletions.
diff --git a/dbt_invoke/internal/_utils.py b/dbt_invoke/internal/_utils.py
@@ -3,6 +3,7 @@
 from pathlib import Path
 import sys
 import platform
+import re
 
 import yaml
 from dbt.task.base import get_nearest_project_dir
@@ -168,7 +169,8 @@ def dbt_ls(
     command = f"dbt {dbt_global_cli_args} ls {dbt_command_cli_args}"
     logger.debug(f'Running command: {command}')
     result = ctx.run(command, hide=hide)
-    result_lines = result.stdout.splitlines()
+    result_stdout = escape_ansi(result.stdout)
+    result_lines = result_stdout.splitlines()
     result_lines_filtered = list()
     for line in result_lines:
         # Because we set the dbt global arg "--log-format json", if
@@ -277,7 +279,8 @@ def dbt_run_operation(
     )
     logger.debug(f'Running command: {command}')
     result = ctx.run(command, hide=hide)
-    result_lines = [json.loads(data) for data in result.stdout.splitlines()]
+    result_stdout = escape_ansi(result.stdout)
+    result_lines = [json.loads(data) for data in result_stdout.splitlines()]
     return result_lines
 
 
@@ -391,6 +394,14 @@ def add_macro(ctx, macro_name, logger=None):
         logger.info(f'Macro "{macro_name}" added to {location.resolve()}')
 
 
+def escape_ansi(line):
+    # Windows can sometime emit Control Sequences in command line outputs
+    # (see https://docs.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences)
+    # The regex filters those out (see https://stackoverflow.com/a/14693789/15202709)
+    ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]')
+    return ansi_escape.sub('', line)
+
+
 class Project:
     """
     A placeholder class for use with get_nearest_project_dir