jq_filters

main.py

#!/usr/bin/env python3
import argparse
import json
import sys
 
 
def jq_filters(d, path=None, text_limit=None):
    """
    Returns all jq filters for the json.
    """
 
    def pprint(v):
        if text_limit:
            if len(str(v)) > text_limit:
                v = str(v)[:text_limit] + "..."
 
        return v
 
    r = {}
 
    if path is None:
        path = [""]
 
    if isinstance(d, dict):
        for k in d.keys():
            r.update(jq_filters(d[k], path=path + [k]))
    elif isinstance(d, list):
        if not d:
            r[".".join(path)] = []
        else:
            for i, e in enumerate(d):
                r.update(jq_filters(e, path=path[:-1] + [path[-1] + f"[{i}]"]))
    else:
        r[".".join(path)] = pprint(d)
 
    return dict(sorted(r.items()))
 
 
def main():
    parser = argparse.ArgumentParser(description="Print all jq filters for json.")
    parser.add_argument("--limit", type=int, default=None)
    parser.add_argument(
        "-c",
        "--compact",
        dest="compact",
        action="store_true",
        default=False,
    )
    parser.add_argument(
        "-i",
        "--input",
        type=argparse.FileType("r"),
        help="input file",
        default=sys.stdin,
    )
    parser.add_argument(
        "-o",
        "--output",
        type=argparse.FileType("w"),
        default=sys.stdout,
        help="output file",
    )
    args = parser.parse_args()
 
    lines = []
    for line in args.input:
        lines.append(line)
 
    res = jq_filters(json.loads("".join(lines)), text_limit=args.limit)
    if not args.compact:
        try:
            print(
                json.dumps(res, indent=2),
                file=args.output,
            )
        except BrokenPipeError:
            pass
    else:
        for k, v in res.items():
            try:
                print(f"{k} = {v}", file=args.output)
            except BrokenPipeError:
                pass
 
 
if __name__ == "__main__":
    main()