pyld command-line tool

pyld.py

#!/usr/bin/env python
#
# Command-line JSON-LD processor based on PyLD
#
# Copyright © 2021-2022 Pierre-Antoine Champin <[email protected]>

import argparse
import json
import sys
from urllib.parse import urljoin, urlsplit
from urllib.request import urlopen
from pyld import jsonld

COMMANDS = ['compact', 'expand', 'toRdf', 'fromRdf', 'frame', 'normalize']

def main():
    parser = argparse.ArgumentParser(description='JSON-LD processor')

    parser.add_argument('command', metavar="COMMAND", choices=COMMANDS,
                        help="|".join(COMMANDS))
    parser.add_argument('input', nargs='?', default='-',
                        help='the file or URL to process (default: STDIN)')
    parser.add_argument('--base', '-b',
                        help='the base IRI used for the input document and the command line arguments')
    parser.add_argument('--context', '-c',
                        help='the context to use for the COMMAND')
    parser.add_argument('--frame', '-f',
                        help="the frame to use (required by 'frame', ignored by other commands)")
    parser.add_argument('--output', '-o',
                        type=argparse.FileType('w', encoding='utf-8'),
                        nargs='?', default='-',
                        help='output file (default: STDOUT)')
    parser.add_argument('--pretty', '-p',
                        action='store_true',
                        help='force pretty (indented) output for JSON (by default on a tty)')
    parser.add_argument('--format', '-F', default='application/n-quads',
                        help='output format for the toRdf command')

    advanced = parser.add_argument_group("advanced options")
    advanced.add_argument('--expandContext',
                          help="context to be used for expansion")
                          # NB: overrides --context if command is expand and both --context and --expandContext are provided
    advanced.add_argument('--noCompactArrays', action='store_true',
                          help="sets the JSON-LD option 'compactArrays' to false")
    # TODO add more options

    args = parser.parse_args()

    if args.input == "-":
        args.input = "file:/dev/stdin"
        if not args.base:
            args.base = 'file:'
    elif urlsplit(args.input).scheme == '':
        args.input = 'file:' + args.input
    
    if args.command == 'compact':
        compact(args)
    elif args.command == 'expand':
        expand(args)
    elif args.command == 'toRdf':
        to_rdf(args)
    elif args.command == 'fromRdf':
        from_rdf(args)
    elif args.command == 'frame':
        frame(args)
    elif args.command == 'normalize':
        normalize(args)
    else:
        eprint(f"{args.command} not implemented")
        exit(-1)
        

def compact(args):
    if args.context is None:
        eprint("--context is required for command 'compact'")
        exit(-2)
    compacted = jsonld.compact(args.input, args.context, options=make_options(args))
    dump_json(args, compacted)

def expand(args):
    options = make_options(args)
    if args.expandContext or args.context:
        options['expandContext'] = args.expandContext or args.context
    expanded = jsonld.expand(args.input, options=options)
    dump_json(args, expanded)

def to_rdf(args):
    options = make_options(args)
    if args.expandContext or args.context:
        options['expandContext'] = args.expandContext or args.context
    rdf = jsonld.to_rdf(args.input, options=options)
    print(rdf, file=args.output)

def from_rdf(args):
    options = make_options(args)
    dataset = loader(args.input, options)['document']
    obj = jsonld.from_rdf(dataset, options)
    if args.context:
        obj = jsonld.compact(obj, args.context, options)
    dump_json(args, obj)

def frame(args):
    if args.frame is None:
        eprint("--frame is required for command 'frame'")
        exit(-3)
    framed = jsonld.frame(args.input, args.frame, options=make_options(args))
    dump_json(args, framed)

def normalize(args):
    options = make_options(args)
    if args.expandContext or args.context:
        options['expandContext'] = args.expandContext or args.context
    rdf = jsonld.normalize(args.input, options=options)
    print(rdf, file=args.output)

def make_options(args):
    opts = {}
    if args.base:
        opts['base'] = args.base
    if args.format:
        opts['format'] = args.format
    if args.noCompactArrays:
        opts['compactArrays'] = false
    return opts

def dump_json(args, data):
    if args.output.isatty() or args.pretty:
        indent='  '
    else:
        indent=None
    json.dump(data, args.output, indent=indent, ensure_ascii=False)

def loader(url: str, options={}):
    #eprint("===", "loader", url, options)
    try:
        resp = urlopen(url, timeout=10)
    except ValueError:
        resp = open(url, 'rb')
    if hasattr(resp, 'getheader'):
        ctype = resp.getheader('content-type')
    else:
        ctype = 'application/ld+json'
    doc = resp.read().decode('utf-8')
    try:
        doc = json.loads(doc)
    except json.JSONDecodeError as err:
        ctype = 'application/x-octet-stream'
    return {
        'document': doc,
        'documentUrl': url,
        'contentType': ctype,
        'contextUrl': None,
    }

def eprint(*args, **kw):
    kw['file'] = sys.stderr
    print(*args, **kw)

jsonld.set_document_loader(loader)

if __name__ == '__main__':
    main()