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 = "respodns" 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) chkgrp = parser.add_mutually_exclusive_group() desc = "enable all checks instead of only the most likely ones" chkgrp.add_argument("--all", action="store_true", help=desc) desc = "enable only the first check" chkgrp.add_argument("--first", action="store_true", help=desc) desc = "enable debugging verbosity" parser.add_argument("--debug", action="store_true", help=desc) a = parser.parse_args(args) checks = [] + chk.first if a.all: checks += chk.checks elif a.first: pass else: checks += chk.likely opts = Options() opts.dry = a.database is None opts.early_stopping = opts.dry opts.progress = a.progress 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): if a.debug: import logging logging.basicConfig(level=logging.DEBUG) run(main(db, a.path, checks, ipinfo, opts), debug=True) else: run(main(db, a.path, checks, ipinfo, 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)