File: tbp.py
   1 #!/usr/bin/python
   2 
   3 # The MIT License (MIT)
   4 #
   5 # Copyright (c) 2026 pacman64
   6 #
   7 # Permission is hereby granted, free of charge, to any person obtaining a copy
   8 # of this software and associated documentation files (the "Software"), to deal
   9 # in the Software without restriction, including without limitation the rights
  10 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11 # copies of the Software, and to permit persons to whom the Software is
  12 # furnished to do so, subject to the following conditions:
  13 #
  14 # The above copyright notice and this permission notice shall be included in
  15 # all copies or substantial portions of the Software.
  16 #
  17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  23 # SOFTWARE.
  24 
  25 
  26 info = '''
  27 tbp [options...] [python expression] [files/URIs...]
  28 
  29 
  30 Transform Bytes with Python runs a python expression on each whole-input,
  31 read as a bytes-type value.
  32 
  33 The expression can use either `v`, `value`, `d`, or `data` for the current
  34 input.
  35 
  36 Input-sources can be either files or web-URIs. When not given any explicit
  37 named sources, the standard input is used. It's even possible to reuse the
  38 standard input using multiple single dashes (-) in the order needed: stdin
  39 is only read once in this case, and kept for later reuse.
  40 
  41 When the expression results in None, the current input is ignored. When the
  42 expression results in a boolean, this determines whether the whole input is
  43 copied/appended back to the standard output, or ignored.
  44 
  45 Options
  46 
  47 All options can start with either a single or a double leading dash:
  48 
  49     -h, -help                      show this help message
  50     -m, -mod, -module, -modules    import modules named in the next argument,
  51                                    where multiple names are comma-separated
  52     -n, -nil, -none, -null         don't read any input & run expression once
  53     -t, -trace, -traceback         turn exceptions into multi-line tracebacks
  54 '''
  55 
  56 
  57 from sys import argv, exit, stderr, stdin, stdout
  58 from time import sleep
  59 from typing import Generator
  60 
  61 
  62 if len(argv) < 2:
  63     print(info.strip(), file=stderr)
  64     exit(1)
  65 if len(argv) > 1 and argv[1] in ('-h', '--h', '-help', '--help'):
  66     print(info.strip())
  67     exit(0)
  68 
  69 
  70 class Skip:
  71     def __call__(self, x):
  72         return isinstance(x, self.__class__)
  73 
  74 skip = Skip()
  75 
  76 def cond(*args):
  77     for i, e in enumerate(args):
  78         if i % 2 == 0 and i < len(args) - 1 and e:
  79             return args[i + 1]
  80 
  81     return args[-1] if len(args) % 2 == 1 else None
  82 
  83 def maybe(f, x = None):
  84     try:
  85         return f(x)
  86     except Exception as _:
  87         return x
  88 
  89 def number(x):
  90     try:
  91         return int(x)
  92     except Exception as _:
  93         pass
  94     try:
  95         return float(x)
  96     except Exception as _:
  97         return x
  98 
  99 def rescue(attempt, fallback = None):
 100     try:
 101         return attempt()
 102     except BrokenPipeError as e:
 103         raise e
 104     except Exception as e:
 105         if callable(fallback):
 106             return fallback(e)
 107         return fallback
 108 
 109 rescued = rescue
 110 
 111 def wait(seconds, result):
 112     t = (int, float)
 113     if (not isinstance(seconds, t)) and isinstance(result, t):
 114         seconds, result = result, seconds
 115     sleep(seconds)
 116     return result
 117 
 118 delay = wait
 119 
 120 def uint_big_endian(src, size, start = 0):
 121     if not isinstance(src, bytes):
 122         return ValueError('can only get unsigned integers from bytes')
 123     if start + size >= len(src):
 124         msg = f'not enough bytes for {8 * size}-bit unsigned integers'
 125         raise ValueError(msg)
 126     return sum(int(src[start + i]) << (8 * i) for i in range(size, 0, -1))
 127 
 128 def uint16_be(src, start = 0):
 129     return uint_big_endian(src, 2, start)
 130 
 131 uint16be = uint16_be
 132 
 133 def uint32_be(src, start = 0):
 134     return uint_big_endian(src, 4, start)
 135 
 136 uint32be = uint32_be
 137 
 138 def uint64_be(src, start = 0):
 139     return uint_big_endian(src, 8, start)
 140 
 141 uint64be = uint64_be
 142 
 143 def uint_little_endian(src, size, start = 0):
 144     if not isinstance(src, bytes):
 145         return ValueError('can only get unsigned integers from bytes')
 146     if start + size >= len(src):
 147         msg = f'not enough bytes for {8 * size}-bit unsigned integers'
 148         raise ValueError(msg)
 149     return sum(int(src[start + i]) << (8 * i) for i in range(size))
 150 
 151 def uint16_le(src, start = 0):
 152     return uint_little_endian(src, 2, start)
 153 
 154 uint16le = uint16_le
 155 
 156 def uint32_le(src, start = 0):
 157     return uint_little_endian(src, 4, start)
 158 
 159 uint32le = uint32_le
 160 
 161 def uint64_le(src, start = 0):
 162     return uint_little_endian(src, 8, start)
 163 
 164 uint64le = uint64_le
 165 
 166 def make_open_read(open):
 167     'Restrict the file-open func to a read-only-binary file-open func.'
 168     def open_read(name):
 169         return open(name, mode='rb')
 170     return open_read
 171 
 172 def fail(msg, code = 1):
 173     print(str(msg), file=stderr)
 174     exit(code)
 175 
 176 def message(msg, result = None):
 177     print(msg, file=stderr)
 178     return result
 179 
 180 msg = message
 181 
 182 def seemsurl(s):
 183     protocols = ('https://', 'http://', 'file://', 'ftp://', 'data:')
 184     return any(s.startswith(p) for p in protocols)
 185 
 186 def tobytes(x):
 187     if isinstance(x, (bytearray, bytes)):
 188         return x
 189     if isinstance(x, (bool, int)):
 190         return bytes([int(x)])
 191     if isinstance(x, float):
 192         return bytes(str(x), encoding='utf-8')
 193     if isinstance(x, str):
 194         return bytes(x, encoding='utf-8')
 195     return bytes(x)
 196 
 197 def tointorbytes(x):
 198     return x if isinstance(x, int) else tobytes(x)
 199 
 200 def adapt_result(x, default):
 201     if x is True:
 202         return default
 203     if x is False:
 204         return None
 205 
 206     if isinstance(x, Skip):
 207         return None
 208 
 209     if callable(x):
 210         return x(default)
 211     return x
 212 
 213 def emit_result(w, x):
 214     if x is None:
 215         return
 216 
 217     if isinstance(x, int):
 218         w.write(tobytes(x))
 219         return
 220 
 221     if isinstance(x, (list, range, set, tuple, Generator)):
 222         for e in x:
 223             w.write(tobytes(e))
 224         return
 225 
 226     w.write(tobytes(x))
 227 
 228 def eval_expr(expr, using):
 229     global v, val, value, d, dat, data
 230     # offer several aliases for the variable with the input bytes
 231     v = val = value = d = dat = data = using
 232     return adapt_result(eval(expr), using)
 233 
 234 
 235 true = True
 236 false = False
 237 nil = none = null = None
 238 
 239 
 240 exec = None
 241 open = make_open_read(open)
 242 
 243 no_input_opts = (
 244     '=', '-n', '--n', '-nil', '--nil', '-none', '--none', '-null', '--null',
 245 )
 246 modules_opts = (
 247     '-m', '--m', '-mod', '--mod', '-module', '--module',
 248     '-modules', '--modules',
 249 )
 250 
 251 args = argv[1:]
 252 load_input = True
 253 expression = None
 254 
 255 while len(args) > 0:
 256     if args[0] == '--':
 257         args = args[1:]
 258         break
 259 
 260     if args[0] in no_input_opts:
 261         load_input = False
 262         args = args[1:]
 263         continue
 264 
 265     if args[0] in no_input_opts:
 266         no_input = True
 267         args = args[1:]
 268         continue
 269 
 270     if args[0] in modules_opts:
 271         try:
 272             if len(args) < 2:
 273                 msg = 'a module name or a comma-separated list of modules'
 274                 raise Exception('expected ' + msg)
 275 
 276             g = globals()
 277             from importlib import import_module
 278             for e in args[1].split(','):
 279                 g[e] = import_module(e)
 280 
 281             g = None
 282             import_module = None
 283             args = args[2:]
 284         except Exception as e:
 285             print(str(e), file=stderr)
 286             exit(1)
 287 
 288         continue
 289 
 290     break
 291 
 292 amp = b'&'
 293 apos = b'\''
 294 lcur = lcurly = b'{'
 295 quot = b'"'
 296 rcur = rcurly = b'}'
 297 utf8bom = b'\xef\xbb\xbf'
 298 bom = {
 299     'utf8': b'\xef\xbb\xbf',
 300     'utf16be': b'\xfe\xff',
 301     'utf16le': b'\xff\xfe',
 302     'utf32be': b'\x00\x00\xfe\xff',
 303     'utf32le': b'\xff\xfe\x00\x00',
 304 }
 305 
 306 if len(args) > 0:
 307     expression = args[0]
 308     args = args[1:]
 309 
 310 if expression is None:
 311     print(info.strip(), file=stderr)
 312     exit(0)
 313 
 314 try:
 315     if not expression or expression == '.':
 316         expression = 'data'
 317     expression = compile(expression, expression, 'eval')
 318 
 319     got_stdin = False
 320     all_stdin = None
 321     dashes = args.count('-')
 322 
 323     data = None
 324 
 325     if not load_input:
 326         emit_result(stdout.buffer, eval_expr(expression, None))
 327         exit(0)
 328 
 329     if any(seemsurl(name) for name in args):
 330         from urllib.request import urlopen
 331 
 332     for name in args:
 333         if name == '-':
 334             if dashes > 1:
 335                 if not got_stdin:
 336                     all_stdin = stdin.buffer.read()
 337                     got_stdin = True
 338                 data = all_stdin
 339             else:
 340                 data = stdin.buffer.read()
 341         elif seemsurl(name):
 342             with urlopen(name) as inp:
 343                 data = inp.read()
 344         else:
 345             with open(name) as inp:
 346                 data = inp.read()
 347         emit_result(stdout.buffer, eval_expr(expression, data))
 348 
 349     if len(args) == 0:
 350         data = stdin.buffer.read()
 351         emit_result(stdout.buffer, eval_expr(expression, data))
 352 except BrokenPipeError:
 353     # quit quietly, instead of showing a confusing error message
 354     stderr.close()
 355     exit(0)
 356 except KeyboardInterrupt:
 357     exit(2)
 358 except Exception as e:
 359     # raise e
 360     s = str(e)
 361     s = s if s else '<generic exception>'
 362     print(s, file=stderr)
 363     exit(1)