Making the b64 binary a lot more flexible. It now allows only parts of the consumed line to be b64 encoded/decoded.

Author: freol35241

README.md

@@ -69,7 +69,7 @@ services:
 
 * **b64**
 
-  Base64 encodes (`--encode`) or decodes (`--decode`) data from STDIN to STDOUT.
+  Base64 encodes (`--encode`) or decodes (`--decode`) data from STDIN to STDOUT. Optinally takes two arguments, the `input_format_specification` and the `output_format_specification` to flexibly allow only parts of the input to be encoded/decoded.
 
 * **jsonify**
 

bin/b64

@@ -5,23 +5,83 @@ Command line utility tool for processing input from stdin. Each line on the
 input stream is base64 encoded with no wrapping and ended with a newline.
 """
 
+# pylint: disable=duplicate-code
+
 import sys
+import logging
+import warnings
 import argparse
 from base64 import b64encode, b64decode
 
+import parse
+
 parser = argparse.ArgumentParser()
 group = parser.add_mutually_exclusive_group(required=True)
 group.add_argument("--encode", action="store_true", default=False)
 group.add_argument("--decode", action="store_true", default=False)
+parser.add_argument(
+    "--log-level", type=lambda level: getattr(logging, level), default=logging.WARNING
+)
+parser.add_argument(
+    "input_specification",
+    type=str,
+    nargs="?",
+    default="{input}",
+    help="Example: '{timestamp} {data}',"
+    "See https://github.com/r1chardj0n3s/parse#format-specification",
+)
+parser.add_argument(
+    "output_specification",
+    type=str,
+    nargs="?",
+    default="{output}",
+    help="Example: '{data}',"
+    "See https://github.com/r1chardj0n3s/parse#format-specification",
+)
 
 args = parser.parse_args()
 
-if args.encode:
-    for line in sys.stdin:
-        sys.stdout.write(b64encode(line.encode()).decode() + "\n")
-        sys.stdout.flush()
+# Setup logger
+logging.basicConfig(
+    format="%(asctime)s %(levelname)s %(name)s %(message)s", level=args.log_level
+)
+logging.captureWarnings(True)
+warnings.filterwarnings("once")
+
+logger = logging.getLogger("b64")
+
+# Compile pattern
+input_pattern = parse.compile(args.input_specification)
+
+for line in sys.stdin:
+    logger.debug(line)
+    res = input_pattern.parse(line.rstrip())
+
+    if not res:
+        logger.error(
+            "Could not parse line: %s according to the input_specification: %s",
+            line,
+            args.input_specification,
+        )
+        continue
+
+    if not "input" in res.named:
+        logger.error(
+            "Could not find the expected named argument 'input' in the input specification: %s",
+            args.input_specification,
+        )
+        continue
+
+    parts = res.named
+
+    _input = parts.pop("input")
+    output = (
+        b64encode(_input.encode()).decode()
+        if args.encode
+        else b64decode(_input.encode()).decode()
+    )
+
+    parts["output"] = output
 
-else:
-    for line in sys.stdin:
-        sys.stdout.write(b64decode(line.encode()).decode())
-        sys.stdout.flush()
+    sys.stdout.write(args.output_specification.format(**parts) + "\n")
+    sys.stdout.flush()

bin/jsonify

@@ -6,6 +6,8 @@ input stream is parsed according to the specification provided by the user
 and assembled into a json object.
 """
 
+# pylint: disable=duplicate-code
+
 import sys
 import json
 import logging