respodns/respodns/ui.py

def ui(program, args):
    from .db import RespoDB
    from .dns import main
    from .ip_info import IpInfoByIpApi
    from .structs import Options
    from argparse import ArgumentParser
    from asyncio import run
    import respodns.checks as chk

    name = "respodns6"
    desc = name + ": test and log DNS records"
    parser = ArgumentParser(name, description=desc)

    desc = "a path to a file containing IPv4 addresses which host DNS servers"
    parser.add_argument("path", metavar="file-path", nargs="+", help=desc)

    parser.add_argument("--database", help="specify database for logging")

    desc = "enable pretty-printing progress to stderr"
    parser.add_argument("--progress", action="store_true", help=desc)

    a = parser.parse_args(args)

    checks = chk.first + chk.likely

    opts = Options()
    opts.dry = a.database is None
    opts.early_stopping = opts.dry
    opts.progress = a.progress

    opts.ipinfo = IpInfoByIpApi("ipinfo_cache.csv")

    if a.database is not None:
        if a.database.startswith("sqlite:"):
            uri = a.database
        else:
            uri = "sqlite:///" + a.database

    def runwrap(db, debug=False):
        if debug:
            import logging
            logging.basicConfig(level=logging.DEBUG)
            run(main(db, a.path, checks, opts), debug=True)
        else:
            run(main(db, a.path, checks, opts))

    if opts.dry:
        runwrap(None)
    else:
        # log to a database.
        db = RespoDB(uri, create=True)
        with db:  # TODO: .open and .close methods for manual invocation.
            with db.execution as execution:  # TODO: clean up this interface.
                opts.execution = execution
                runwrap(db)