File: hipy.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 hipy [options...] [python expression] [files/URIs...] 28 29 30 Highlight using PYthon colors lines by running the Python expression given on 31 each line, and using its results as booleans. Inputs are encoded as UTF-8. If 32 no expression is given, all lines are highlighted by default. Highlighting 33 happens via simple ANSI-codes. 34 35 Carriage-returns are always ignored in lines, as well as any UTF-8-BOM on the 36 first line of each input. 37 38 The expression can use either `l` or `line` for the current line, and `i` as 39 a 0-based line counter which keeps growing even across input-sources, when 40 given multiple inputs. Also available is `n`, a 1-based line counter which 41 otherwise works the same way. 42 43 Each line is automatically parsed as JSON: when successful, the parsed line 44 is available to the expression as `v`, or `value`. You can check failure to 45 parse JSON by checking if `v` or `value` is of type Skip, since None can be 46 the result of successfully parsing a null JSON value. 47 48 Input-sources can be either files or web-URIs. When not given any explicit 49 named sources, the standard input is used. A single dash also means the 50 standard input. 51 52 When the expression results in None, the current input line is ignored, 53 allowing expressions to also act as convenient line filters. 54 55 56 Options 57 58 All options can start with either a single or a double leading dash: 59 60 -h, -help show this help message 61 -m, -mod, -module, -modules import modules named in the next argument, 62 where multiple names are comma-separated 63 -t, -trace, -traceback turn exceptions into multi-line tracebacks 64 65 66 Examples 67 68 ''' 69 70 71 from itertools import islice, zip_longest 72 from json import dump, dumps, loads 73 from math import isinf, isnan 74 from re import compile as compile_uncached, IGNORECASE 75 from sys import argv, exit, stderr, stdin, stdout 76 from time import localtime, strftime 77 from typing import Iterable 78 79 80 if len(argv) > 1 and argv[1] in ('-h', '--h', '-help', '--help'): 81 print(info.strip()) 82 exit(0) 83 84 85 def style_lines(src, expr): 86 # `comprehension` expressions seem to ignore local variables: even 87 # lambda-based workaround-attempts fail to make needed values like 88 # the current line available to such expressions 89 global i, n, fn, l, line, v, val, value 90 global items, fields, words, tsv 91 92 fn = 1 93 94 for l in src: 95 l = l.rstrip('\r\n').rstrip('\n') 96 if fn == 1: 97 l = l.lstrip('\xef\xbb\xbf') 98 99 line = l 100 items = fields = words = line.split() 101 tsv = line.split('\t') 102 103 try: 104 v = val = value = loads(l) 105 except Exception as _: 106 v = val = value = Skip() 107 108 ok = eval(expr) 109 if callable(ok) and (not isinstance(ok, Skip)): 110 ok = ok(line) 111 if isinstance(ok, BaseException): 112 raise ok 113 114 i += 1 115 n += 1 116 fn += 1 117 118 if (ok is None) or isinstance(ok, Skip): 119 continue 120 121 if not ok: 122 print(line, flush=flushed) 123 continue 124 125 if not ('\x1b[0m' in line): 126 print(f'\x1b[7m{line}\x1b[0m', flush=flushed) 127 continue 128 129 print('\x1b[7m', end='') 130 for s in line.split('\x1b[0m'): 131 print(s, end='') 132 print('\x1b[0m\x1b[7m', end='') 133 print('\x1b[0m', flush=flushed) 134 135 def fail(msg, code = 1): 136 print(str(msg), file=stderr) 137 exit(code) 138 139 def make_open_utf8(open): 140 def open_utf8_readonly(path): 141 return open(path, encoding='utf-8') 142 return open_utf8_readonly 143 144 def seemsurl(path): 145 protocols = ('https://', 'http://', 'file://', 'ftp://', 'data:') 146 return any(path.startswith(p) for p in protocols) 147 148 class Skip: 149 def __call__(self, x): 150 return isinstance(x, self.__class__) 151 152 skip = Skip() 153 154 def cases(x, *args): 155 ''' 156 Simulate a switch statement on a value, using matches/result pairs from 157 the arguments given; when given an even number of extra args, None is 158 used as a final fallback result; when given an odd number of extra args, 159 the last argument is used as a final `default` value, if needed. 160 ''' 161 162 for i in range(0, len(args) - len(args) % 2, 2): 163 test, res = args[i], args[i+1] 164 if isinstance(test, (list, tuple)) and (x in test): 165 return res 166 if isinstance(test, float) and isnan(test) and isnan(x): 167 return res 168 if x == test: 169 return res 170 return None if len(args) % 2 == 0 else args[-1] 171 172 switch = cases 173 174 def chunk(items, chunk_size): 175 'Break iterable into chunks, each with up to the item-count given.' 176 177 if isinstance(items, str): 178 n = len(items) 179 while n >= chunk_size: 180 yield items[:chunk_size] 181 items = items[chunk_size:] 182 n -= chunk_size 183 if n > 0: 184 yield items 185 return 186 187 if not isinstance(chunk_size, int): 188 raise Exception('non-integer chunk-size') 189 if chunk_size < 1: 190 raise Exception('non-positive chunk-size') 191 192 it = iter(items) 193 while True: 194 head = tuple(islice(it, chunk_size)) 195 if not head: 196 return 197 yield head 198 199 chunked = chunk 200 201 # re_cache is used by custom func compile to cache previously-compiled 202 # regular-expressions, which makes them quicker to (re)use in formulas 203 re_cache = {} 204 205 def re_compile(expr, flags = 0): 206 'Speed-up using regexes across lines, by avoiding recompilations.' 207 208 if flags in re_cache: 209 cache = re_cache[flags] 210 else: 211 cache = {} 212 re_cache[flags] = cache 213 if expr in cache: 214 return cache[expr] 215 216 pat = compile_uncached(expr, flags) 217 cache[expr] = pat 218 return pat 219 220 def icompile(expr): 221 return re_compile(expr, IGNORECASE) 222 223 def cond(*args): 224 for i, e in enumerate(args): 225 if i % 2 == 0 and i < len(args) - 1 and e: 226 return args[i + 1] 227 228 return args[-1] if len(args) % 2 == 1 else None 229 230 def denan(x, fallback = None): 231 'Replace floating-point NaN with the alternative value given.' 232 return x if not (isinstance(x, float) and isnan(x)) else fallback 233 234 def denil(*args): 235 'Avoid None values, if possible: first value which isn\'t None wins.' 236 for e in args: 237 if e != None: 238 return e 239 return None 240 241 denone = denull = denil 242 243 def dive(into, using): 244 'Depth-first recursive caller for 1-input functions.' 245 246 if callable(into): 247 into, using = using, into 248 249 def rec(v): 250 if isinstance(v, dict): 251 return {k: rec(v) for k, v in v.items()} 252 if isinstance(v, Iterable) and not isinstance(v, str): 253 return [rec(v) for v in v] 254 return using(v) 255 256 return rec(into) 257 258 def divekeys(into, using): 259 'Depth-first recursive caller for 2-input funcs which rename dict keys.' 260 261 if callable(into): 262 into, using = using, into 263 264 def rec(v): 265 if isinstance(v, dict): 266 return {using(k): rec(v) for k, v in v.items()} 267 if isinstance(v, Iterable) and not isinstance(v, str): 268 return [rec(v) for i, v in enumerate(v)] 269 return v 270 271 return rec(None, into) 272 273 def divekv(into, using, using2 = None): 274 'Depth-first recursive caller for 2-input functions.' 275 276 if using2 is None: 277 if callable(into): 278 into, using = using, into 279 else: 280 if not callable(using2): 281 into, using, using2 = using2, into, using 282 283 def rec(k, v): 284 if isinstance(v, dict): 285 return {k: rec(k, v) for k, v in v.items()} 286 if isinstance(v, Iterable) and not isinstance(v, str): 287 return [rec(i, v) for i, v in enumerate(v)] 288 return using(k, v) 289 290 def rec2(k, v): 291 if isinstance(v, dict): 292 return {str(using(k, v)): rec2(k, v) for k, v in v.items()} 293 if isinstance(v, Iterable) and not isinstance(v, str): 294 return [rec2(i, v) for i, v in enumerate(v)] 295 return using2(k, v) 296 297 return rec(None, into) if using2 is None else rec2(None, into) 298 299 kvdive = divekv 300 301 def drop(src, *what): 302 if isinstance(src, str): 303 for s in what: 304 src = src.replace(s, '') 305 return src 306 307 def kdrop(src, what): 308 return {k: v for (k, v) in src.items() if not (k in what)} 309 310 if isinstance(src, dict): 311 return kdrop(src, set(what)) 312 313 if isinstance(src, Iterable): 314 what = set(what) 315 return [kdrop(e, what) for e in src if isinstance(e, dict)] 316 317 return None 318 319 dropped = drop 320 321 def flat(*args): 322 'Turn arbitrarily-nested values/sequences into a single flat sequence.' 323 324 def rec(x): 325 if x is None: 326 return 327 328 if isinstance(x, dict): 329 yield from rec(x.values()) 330 331 if isinstance(x, str): 332 yield x 333 return 334 335 if isinstance(x, Iterable): 336 for e in x: 337 yield from rec(e) 338 return 339 340 yield x 341 342 for x in args: 343 yield from rec(x) 344 345 flatten = flattened = flat 346 347 def identity(x): 348 return x 349 350 def iota(n): 351 return range(1, n + 1) 352 353 def join(x, *y): 354 'Join values into a string, or make a dict from keys and values.' 355 356 if len(y) == 0: 357 return ' '.join(str(v) for v in x) 358 if isinstance(x, str): 359 return x.join(str(v) for v in y) 360 if len(y) == 1 and isinstance(y[0], str): 361 return y[0].join(str(v) for v in x) 362 if len(y) == 1 and isinstance(y[0], (list, range, set, tuple, Generator)): 363 return {k: v for k, v in zip_longest(x, y[0]) if not (k is None)} 364 if isinstance(y, (list, range, set, tuple, Generator)): 365 return {k: v for k, v in zip_longest(x, y) if not (k is None)} 366 return {k: y for k in x} 367 368 joined = join 369 370 def maybe(f, x = None): 371 try: 372 return f(x) 373 except Exception as _: 374 return x 375 376 def number(x): 377 try: 378 return int(x) 379 except Exception as _: 380 pass 381 try: 382 return float(x) 383 except Exception as _: 384 return x 385 386 def pick(src, *keys): 387 if isinstance(src, dict): 388 return {k: src.get(k, None) for k in keys} 389 return [{k: e.get(k, None) for k in keys} for e in src if isinstance(e, dict)] 390 391 picked = pick 392 393 def plain(s): 394 'Ignore all ANSI-style sequences in a string.' 395 return re_compile('''\x1b\\[([0-9;]+m|[0-9]*[A-HJKST])''').sub('', s) 396 397 def predicate(x): 398 'Helps various higher-order funcs, by standardizing `predicate` values.' 399 if callable(x): 400 return x 401 if not isinstance(x, float): 402 return lambda y: x == y 403 if isnan(x): 404 return lambda y: isinstance(y, float) and isnan(y) 405 if isinf(x): 406 return lambda y: isinstance(y, float) and isinf(y) 407 408 def rescue(attempt, fallback = None): 409 try: 410 return attempt() 411 except Exception as e: 412 if callable(fallback) and (not isinstance(fallback, Skip)): 413 return fallback(e) 414 return fallback 415 416 rescued = rescue 417 418 def retype(x): 419 'Try to narrow the type of the value given.' 420 421 if isinstance(x, float): 422 n = int(x) 423 return n if float(n) == x else x 424 425 if not isinstance(x, str): 426 return x 427 428 try: 429 return loads(x) 430 except Exception as _: 431 pass 432 433 try: 434 return int(x) 435 except Exception as _: 436 pass 437 438 try: 439 return float(x) 440 except Exception as _: 441 pass 442 443 return x 444 445 autocast = autocasted = mold = molded = recast = recasted = remold = retype 446 remolded = retyped = retype 447 448 def json0(x): 449 if isinstance(x, (range, set, Generator)): 450 x = tuple(x) 451 return dumps(x, separators=(',', ':'), allow_nan=False, indent=None) 452 453 j0 = json0 454 455 def typeof(x): 456 return { 457 type(None): 'null', 458 bool: 'boolean', 459 dict: 'object', 460 float: 'number', int: 'number', 461 str: 'string', 462 list: 'array', tuple: 'array', 463 }.get(type(x), 'other') 464 465 jstype = typeof 466 467 def after(x, what): 468 i = x.find(what) 469 return '' if i < 0 else x[i+len(what):] 470 471 def afterlast(x, what): 472 i = x.rfind(what) 473 return '' if i < 0 else x[i+len(what):] 474 475 afterfinal = afterlast 476 477 def before(x, what): 478 i = x.find(what) 479 return x if i < 0 else x[:i] 480 481 def beforelast(x, what): 482 i = x.rfind(what) 483 return x if i < 0 else x[:i] 484 485 beforefinal = beforelast 486 487 def since(x, what): 488 i = x.find(what) 489 return '' if i < 0 else x[i:] 490 491 def sincelast(x, what): 492 i = x.rfind(what) 493 return '' if i < 0 else x[i:] 494 495 sincefinal = sincelast 496 497 def until(x, what): 498 i = x.find(what) 499 return x if i < 0 else x[:i+len(what)] 500 501 def untilfinal(x, what): 502 i = x.rfind(what) 503 return x if i < 0 else x[:i+len(what)] 504 505 untillast = untilfinal 506 507 def message(msg, result = None): 508 print(msg, file=stderr) 509 return result 510 511 msg = message 512 513 # seen is used by func `once` to remember previously-given values 514 seen = set() 515 516 def once(x, key = None): 517 k = key(x) if key else x 518 if k in seen: 519 return None 520 seen.add(k) 521 return x 522 523 dedup = unique = once 524 525 def utf8(x): 526 try: 527 if isinstance(x, str): 528 x = x.encode('utf-8') 529 return str(x, 'utf-8') 530 except Exception as _: 531 return None 532 533 def ymdhms(when = None): 534 fmt = f'%Y-%m-%d %H:%M:%S' 535 if isinstance(when, (float, int)): 536 return strftime(fmt, localtime(float(when))) 537 if isinstance(when, tuple): 538 return strftime(fmt, when) 539 return strftime(fmt, localtime()) 540 541 542 amp = '&' 543 apos = '\'' 544 lcur = lcurly = '{' 545 quot = '"' 546 rcur = rcurly = '}' 547 548 true = True 549 false = False 550 nil = none = null = None 551 552 553 exec = None 554 open_utf8 = make_open_utf8(open) 555 open = open_utf8 556 557 modules_opts = ( 558 '-m', '--m', '-mod', '--mod', '-module', '--module', 559 '-modules', '--modules', 560 ) 561 trace_opts = ('-t', '--t', '-trace', '--trace', '-traceback', '--traceback') 562 563 args = argv[1:] 564 if any(seemsurl(e) for e in args): 565 from io import TextIOWrapper 566 from urllib.request import urlopen 567 568 tracing = False 569 570 while len(args) > 0: 571 if args[0] == '-': 572 break 573 574 if args[0] == '--': 575 args = args[1:] 576 break 577 578 if args[0] in modules_opts: 579 try: 580 if len(args) < 2: 581 msg = 'a module name or a comma-separated list of modules' 582 raise Exception('expected ' + msg) 583 584 g = globals() 585 from importlib import import_module 586 for e in args[1].split(','): 587 g[e] = import_module(e) 588 589 g = None 590 import_module = None 591 args = args[2:] 592 except Exception as e: 593 fail(e, 1) 594 595 continue 596 597 if args[0] in trace_opts: 598 tracing = True 599 args = args[1:] 600 continue 601 602 if args[0].startswith('-'): 603 fail(f'hipy: unsupported option {args[0]}', 1) 604 exit(1) 605 606 break 607 608 609 # ensure live-lines output, unless stdout is being saved into a file 610 flushed = stdout.isatty() or (not stdout.seekable()) 611 612 expr = '.' 613 if len(args) > 0: 614 expr = args[0] 615 args = args[1:] 616 617 if expr == '.': 618 expr = 'True' 619 620 try: 621 expr = compile(expr, expr, mode='eval') 622 compile = None 623 624 i = 0 625 n = 1 626 v = val = value = Skip() 627 628 items = fields = words = [] 629 tsv = [] 630 631 if len(args) == 0: 632 style_lines(stdin, expr) 633 634 if args.count('-') > 1: 635 msg = 'reading from `-` (standard input) more than once not allowed' 636 raise ValueError(msg) 637 638 for path in args: 639 if path == '-': 640 style_lines(stdin, expr) 641 continue 642 643 if seemsurl(path): 644 with urlopen(path) as inp: 645 with TextIOWrapper(inp, encoding='utf-8') as txt: 646 style_lines(txt, expr) 647 continue 648 649 with open_utf8(path) as txt: 650 style_lines(txt, expr) 651 except BrokenPipeError: 652 # quit quietly, instead of showing a confusing error message 653 stderr.close() 654 exit(0) 655 except KeyboardInterrupt: 656 exit(2) 657 except Exception as e: 658 if tracing: 659 raise e 660 fail(e, 1)