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     if len(args) == 0:
  78         return None
  79 
  80     for i, e in enumerate(args):
  81         if i % 2 == 0 and i < len(args) - 1 and e:
  82             return args[i + 1]
  83 
  84     return args[-1] if len(args) % 2 == 1 else None
  85 
  86 def maybe(f, x):
  87     try:
  88         return f(x)
  89     except Exception as _:
  90         return x
  91 
  92 def number(x):
  93     try:
  94         return int(x)
  95     except Exception as _:
  96         pass
  97     try:
  98         return float(x)
  99     except Exception as _:
 100         return x
 101 
 102 def rescue(attempt, fallback = None):
 103     try:
 104         return attempt()
 105     except BrokenPipeError as e:
 106         raise e
 107     except Exception as e:
 108         if callable(fallback):
 109             return fallback(e)
 110         return fallback
 111 
 112 rescued = rescue
 113 
 114 def wait(seconds, result):
 115     t = (int, float)
 116     if (not isinstance(seconds, t)) and isinstance(result, t):
 117         seconds, result = result, seconds
 118     sleep(seconds)
 119     return result
 120 
 121 delay = wait
 122 
 123 def uint_big_endian(src, size, start = 0):
 124     if not isinstance(src, bytes):
 125         return ValueError('can only get unsigned integers from bytes')
 126     if start + size >= len(src):
 127         msg = f'not enough bytes for {8 * size}-bit unsigned integers'
 128         raise ValueError(msg)
 129     return sum(int(src[start + i]) << (8 * i) for i in range(size, 0, -1))
 130 
 131 def uint16_be(src, start = 0):
 132     return uint_big_endian(src, 2, start)
 133 
 134 uint16be = uint16_be
 135 
 136 def uint32_be(src, start = 0):
 137     return uint_big_endian(src, 4, start)
 138 
 139 uint32be = uint32_be
 140 
 141 def uint64_be(src, start = 0):
 142     return uint_big_endian(src, 8, start)
 143 
 144 uint64be = uint64_be
 145 
 146 def uint_little_endian(src, size, start = 0):
 147     if not isinstance(src, bytes):
 148         return ValueError('can only get unsigned integers from bytes')
 149     if start + size >= len(src):
 150         msg = f'not enough bytes for {8 * size}-bit unsigned integers'
 151         raise ValueError(msg)
 152     return sum(int(src[start + i]) << (8 * i) for i in range(size))
 153 
 154 def uint16_le(src, start = 0):
 155     return uint_little_endian(src, 2, start)
 156 
 157 uint16le = uint16_le
 158 
 159 def uint32_le(src, start = 0):
 160     return uint_little_endian(src, 4, start)
 161 
 162 uint32le = uint32_le
 163 
 164 def uint64_le(src, start = 0):
 165     return uint_little_endian(src, 8, start)
 166 
 167 uint64le = uint64_le
 168 
 169 def make_open_read(open):
 170     'Restrict the file-open func to a read-only-binary file-open func.'
 171     def open_read(name):
 172         return open(name, mode='rb')
 173     return open_read
 174 
 175 def fail(msg, code = 1):
 176     print(str(msg), file=stderr)
 177     exit(code)
 178 
 179 def message(msg, result = None):
 180     print(msg, file=stderr)
 181     return result
 182 
 183 msg = message
 184 
 185 def seemsurl(s):
 186     protocols = ('https://', 'http://', 'file://', 'ftp://', 'data:')
 187     return any(s.startswith(p) for p in protocols)
 188 
 189 def tobytes(x):
 190     if isinstance(x, (bytearray, bytes)):
 191         return x
 192     if isinstance(x, (bool, int)):
 193         return bytes([int(x)])
 194     if isinstance(x, float):
 195         return bytes(str(x), encoding='utf-8')
 196     if isinstance(x, str):
 197         return bytes(x, encoding='utf-8')
 198     return bytes(x)
 199 
 200 def tointorbytes(x):
 201     return x if isinstance(x, int) else tobytes(x)
 202 
 203 def adapt_result(x, default):
 204     if x is True:
 205         return default
 206     if x is False:
 207         return None
 208 
 209     if isinstance(x, Skip):
 210         return None
 211 
 212     if callable(x):
 213         return x(default)
 214     return x
 215 
 216 def emit_result(w, x):
 217     if x is None:
 218         return
 219 
 220     if isinstance(x, int):
 221         w.write(tobytes(x))
 222         return
 223 
 224     if isinstance(x, (list, range, set, tuple, Generator)):
 225         for e in x:
 226             w.write(tobytes(e))
 227         return
 228 
 229     w.write(tobytes(x))
 230 
 231 def eval_expr(expr, using):
 232     global v, val, value, d, dat, data
 233     # offer several aliases for the variable with the input bytes
 234     v = val = value = d = dat = data = using
 235     return adapt_result(eval(expr), using)
 236 
 237 
 238 nil = none = null = None
 239 
 240 
 241 exec = None
 242 open = make_open_read(open)
 243 
 244 no_input_opts = (
 245     '=', '-n', '--n', '-nil', '--nil', '-none', '--none', '-null', '--null',
 246 )
 247 string_opts = ('-s', '--s', '-str', '--str', '-string', '--string')
 248 modules_opts = (
 249     '-m', '--m', '-mod', '--mod', '-module', '--module',
 250     '-modules', '--modules',
 251 )
 252 
 253 args = argv[1:]
 254 load_input = True
 255 string_input = False
 256 expression = None
 257 
 258 while len(args) > 0:
 259     if args[0] == '--':
 260         args = args[1:]
 261         break
 262 
 263     if args[0] in no_input_opts:
 264         load_input = False
 265         args = args[1:]
 266         continue
 267 
 268     if args[0] in string_opts:
 269         string_input = True
 270         args = args[1:]
 271         continue
 272 
 273     if args[0] in no_input_opts:
 274         no_input = True
 275         args = args[1:]
 276         continue
 277 
 278     if args[0] in modules_opts:
 279         try:
 280             if len(args) < 2:
 281                 msg = 'a module name or a comma-separated list of modules'
 282                 raise Exception('expected ' + msg)
 283 
 284             g = globals()
 285             from importlib import import_module
 286             for e in args[1].split(','):
 287                 g[e] = import_module(e)
 288 
 289             g = None
 290             import_module = None
 291             args = args[2:]
 292         except Exception as e:
 293             fail(e, 1)
 294 
 295         continue
 296 
 297     break
 298 
 299 dquo = dquote = '"' if string_input else b'"'
 300 empty = '' if string_input else b''
 301 lcurly = '{' if string_input else b'{'
 302 rcurly = '}' if string_input else b'}'
 303 squo = squote = '\'' if string_input else b'\''
 304 utf8bom = '\xef\xbb\xbf' if string_input else b'\xef\xbb\xbf'
 305 if string_input:
 306     bom = {
 307         'utf8': '\xef\xbb\xbf',
 308         'utf16be': '\xfe\xff',
 309         'utf16le': '\xff\xfe',
 310         'utf32be': '\x00\x00\xfe\xff',
 311         'utf32le': '\xff\xfe\x00\x00',
 312     }
 313 else:
 314     bom = {
 315         'utf8': b'\xef\xbb\xbf',
 316         'utf16be': b'\xfe\xff',
 317         'utf16le': b'\xff\xfe',
 318         'utf32be': b'\x00\x00\xfe\xff',
 319         'utf32le': b'\xff\xfe\x00\x00',
 320     }
 321 
 322 if len(args) > 0:
 323     expression = args[0]
 324     args = args[1:]
 325 
 326 if expression is None:
 327     print(info.strip(), file=stderr)
 328     exit(0)
 329 
 330 try:
 331     if not expression or expression == '.':
 332         expression = 'data'
 333     expression = compile(expression, expression, 'eval')
 334 
 335     got_stdin = False
 336     all_stdin = None
 337     dashes = args.count('-')
 338 
 339     data = None
 340 
 341     if not load_input:
 342         emit_result(stdout.buffer, eval_expr(expression, None))
 343         exit(0)
 344 
 345     if any(seemsurl(name) for name in args):
 346         from urllib.request import urlopen
 347 
 348     for name in args:
 349         if name == '-':
 350             if dashes > 1:
 351                 if not got_stdin:
 352                     all_stdin = stdin.buffer.read()
 353                     got_stdin = True
 354                 data = all_stdin
 355             else:
 356                 data = stdin.buffer.read()
 357 
 358             if string_input:
 359                 data = s = str(data, encoding='utf-8')
 360         elif seemsurl(name):
 361             with urlopen(name) as inp:
 362                 data = inp.read()
 363         else:
 364             with open(name) as inp:
 365                 data = inp.read()
 366 
 367         if string_input:
 368             data = s = str(data, encoding='utf-8')
 369         emit_result(stdout.buffer, eval_expr(expression, data))
 370 
 371     if len(args) == 0:
 372         data = stdin.buffer.read()
 373         if string_input:
 374             data = s = str(data, encoding='utf-8')
 375         emit_result(stdout.buffer, eval_expr(expression, data))
 376 except BrokenPipeError:
 377     # quit quietly, instead of showing a confusing error message
 378     stderr.close()
 379     exit(0)
 380 except KeyboardInterrupt:
 381     exit(2)
 382 except Exception as e:
 383     s = str(e)
 384     s = s if s else '<generic exception>'
 385     print(s, file=stderr)
 386     exit(1)