python类install()的实例源码

def run_generate_file_step():
    item = GenerateFile()
    coloredlogs.install(logging.DEBUG)
    item.loop()

def run_hash_file_step():
    step = HashFile()
    coloredlogs.install(logging.DEBUG)
    step.loop()

def run_remove_file_step():
    step = RemoveFile()
    coloredlogs.install(logging.DEBUG)
    step.loop()

def build_log_file(cfg):
    FORMAT="%(asctime)s;%(levelname)s|%(message)s"
    DATEF="%H-%M-%S"
    logging.basicConfig(formatter=FORMAT, level=logging.DEBUG)
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)

    fh = logging.FileHandler(filename=os.path.join(cfg['outfolder'], 'logfile'+time.strftime("%m-%d")+'.log'))
    fh.setLevel(logging.DEBUG)
    formatter = logging.Formatter("%(asctime)s;%(levelname)s|%(message)s", "%H:%M:%S")
    fh.setFormatter(formatter)
    logger.addHandler(fh)

    LEVEL_STYLES = dict(
        debug=dict(color='magenta'),
        info=dict(color='green'),
        verbose=dict(),
        warning=dict(color='blue'),
        error=dict(color='yellow'),
        critical=dict(color='red',bold=True))
    coloredlogs.install(level=logging.DEBUG, fmt=FORMAT, datefmt=DATEF, level_styles=LEVEL_STYLES)


    args_dict = cfg
    sorted_args = sorted(args_dict.items(), key=operator.itemgetter(0))
    logging.info('######################################################')
    logging.info('# --Configurable Parameters In this Model--')
    for name, val in sorted_args:
        logging.info("# " + name + ":\t" + str(get_list_name(val)))
    logging.info('######################################################')

def info(message, showtime=True, title=True):
        """Info level message"""

        level = 'INFO'
        if True == showtime:
            coloredlogs.install(level=level, fmt='[%(asctime)s] %(levelname)s : %(message)s')
        else:
            coloredlogs.install(level=level, fmt='%(message)s')
        logger = Logger.log(level);
        logger.info(message);
        pass

def critical(message, showtime=True, title=True):
        # type: (object, object, object) -> object
        """Critical level message"""

        level = 'CRITICAL'
        if True == showtime:
            coloredlogs.install(level=level, fmt='[%(asctime)s] %(levelname)s : %(message)s')
        else:
            coloredlogs.install(level=level, fmt='%(message)s')
        logger = Logger.log(level);
        sys.exit(logger.critical(message));

def debug(message, showtime=True):
        """Debug level message"""

        level = 'DEBUG'
        if True == showtime:
            coloredlogs.install(level=level, fmt='[%(asctime)s] %(levelname)s : %(message)s')
        else:
            coloredlogs.install(level=level, fmt='%(message)s')
        logger = Logger.log(level);
        logger.debug(message);
        pass

def main(log_level):
    '''
    Pilot Jenkins from the command line.
    '''
    coloredlogs.install(
        fmt='%(levelname)s %(message)s',
        level=log_level
    )

def execute_plural_server():  # pragma: no cover
    parser = argparse.ArgumentParser(
        prog='plural-server',
        description='ZMQ Reply Server that executes queries')

    parser.add_argument(
        '-b', '--reply-bind-addr',
        default='tcp://*:6000',
        help='ZMQ address to bind to'
    )

    parser.add_argument(
        '--color',
        action='store_true',
        default=True,
        help='colored logs'
    )
    parser.add_argument(
        '-l', '--log-level',
        default='DEBUG',
        choices=('DEBUG', 'INFO', 'WARNING')
    )

    args = parser.parse_args()
    server = GraphServer()
    if args.color:
        coloredlogs.install(level=args.log_level)

    logging.warning('EXPERIMENTAL FEATURE: server')
    try:
        server.run(args.reply_bind_addr)
    except KeyboardInterrupt:
        server.stop()
        sys.stdout.write("\r")
        sys.stderr.write("\r")
        sys.stdout.flush()
        sys.stderr.flush()

def main():
    webwx = MyWXBot()

    logger = logging.getLogger(__name__)
    import coloredlogs
    coloredlogs.install(level='DEBUG')


    webwx.start()

def invoke_by_args():
  import argparse
  import asyncio
  import coloredlogs
  import inspect
  import pprint
  coloredlogs.install(fmt='[%(levelname).1s %(asctime)s %(module)s:%(lineno)d] %(message)s',
                      datefmt='%y%m%d %H:%M:%S')
  parser = argparse.ArgumentParser()
  subparsers = parser.add_subparsers(dest='')
  for name, method in _methods.items():
    subparser = subparsers.add_parser(name)
    argcount = method.__code__.co_argcount
    num_defaults = len(method.__defaults__) if method.__defaults__ else 0
    argoffset = argcount - num_defaults
    for index, argname in enumerate(method.__code__.co_varnames[:argcount]):
      if index < argoffset:
        subparser.add_argument(argname, type=method.__annotations__[argname])
      elif argname in method.__annotations__:
        subparser.add_argument(argname,
                               type=method.__annotations__[argname],
                               nargs='?',
                               default=method.__defaults__[index - argoffset])
  args = parser.parse_args(options.leftovers)
  name = getattr(args, '')
  delattr(args, '')
  if not name:
    parser.print_help()
  else:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(db.init())
    try:
      result = _methods[name](**vars(args))
      if inspect.iscoroutine(result):
        result = loop.run_until_complete(result)
      if options.pretty:
        print_func = pprint.pprint
      else:
        print_func = lambda x: print(x) if x is not None else None
      if hasattr(result, '__aiter__'):
        async def aloop():
          async for entry in result:
            print_func(entry)

        loop.run_until_complete(aloop())
      else:
        print_func(result)
    except KeyboardInterrupt:
      pass
    finally:
      loop.set_exception_handler(lambda loop, context: None)

def main(args=None):
    if args is None:
        args = sys.argv[1:]

    cmd_name, cmd_args = parseopts(args)
    if cmd_name == '--version':
        print(aetros.const.__version__)
        sys.exit(0)

    from aetros.commands.ApiCommand import ApiCommand
    from aetros.commands.PushJobCommand import PushJobCommand
    from aetros.commands.PullJobCommand import PullJobCommand
    from aetros.commands.ServerCommand import ServerCommand
    from aetros.commands.PredictCommand import PredictCommand
    from aetros.commands.PredictionServerCommand import PredictionServerCommand
    from aetros.commands.StartCommand import StartCommand
    from aetros.commands.StartSimpleCommand import StartSimpleCommand
    from aetros.commands.RunCommand import RunCommand
    from aetros.commands.AddCommand import AddCommand
    from aetros.commands.InitCommand import InitCommand
    from aetros.commands.IdCommand import IdCommand
    from aetros.commands.GPUCommand import GPUCommand
    from aetros.commands.AuthenticateCommand import AuthenticateCommand

    commands_dict = {
        'start': StartCommand,
        'start-simple': StartSimpleCommand,
        'authenticate': AuthenticateCommand,
        'predict': PredictCommand,
        'prediction-server': PredictionServerCommand,
        'server': ServerCommand,
        'run': RunCommand,
        'api': ApiCommand,
        'id': IdCommand,
        'push-job': PushJobCommand,
        'pull-job': PullJobCommand,
        'add': AddCommand,
        'init': InitCommand,
        'gpu': GPUCommand,
    }

    if cmd_name not in commands_dict:
        print(("Command %s not found" % (cmd_name,)))
        sys.exit(1)

    level = 'INFO'
    if '--debug' in args or os.getenv('DEBUG') == '1':
        level = 'DEBUG'

    atty = None
    if '1' == os.getenv('AETROS_ATTY'):
        atty = True

    import coloredlogs
    logger = logging.getLogger('aetros-'+cmd_name)
    coloredlogs.install(level=level, logger=logger, isatty=atty)
    command = commands_dict[cmd_name](logger)

    code = command.main(cmd_args)
    sys.exit(code)

def main():
    coloredlogs.install(fmt='%(message)s')

    if os.geteuid() == 0:
        logger.error("Do not run this as root")
        sys.exit(1)

    config = cfg.Config()
    config.create()
    config.read()
    config.verify()

    version = 'bridgy %s' % __version__
    args = docopt(__doc__, version=version)

    if not tmux.is_installed():
        if args ['--tmux'] or config.dig('ssh', 'tmux'):
            logger.warn("Tmux not installed. Cannot support split screen.")
        args['--tmux'] = False

    if args['-v']:
        coloredlogs.install(fmt='%(message)s', level='DEBUG')

    if args['-d']:
        args['-v'] = True
        coloredlogs.install(fmt='%(message)s', level='DEBUG')
        logger.warn("Performing dry run, no actions will be taken.")

    # why doesn't docopt pick up on this?
    if args['-t']:
        args['--tmux'] = True

    if args['--version']:
        logger.info(version)
        sys.exit(0)

    opts = {
        'ssh': ssh_handler,
        'mount': mount_handler,
        'list-mounts': list_mounts_handler,
        'list-inventory': list_inventory_handler,
        'unmount': unmount_handler,
        'update': update_handler,
        'run': run_handler,
    }

    for opt, handler in list(opts.items()):
        if args[opt]:
            try:
                handler(args, config)
            except utils.UnsupportedPlatform as ex:
                logger.error(ex.message)
                sys.exit(1)

def main():
    """Command line interface for the ``qpass`` program."""
    # Initialize logging to the terminal.
    coloredlogs.install()
    # Prepare for command line argument parsing.
    action = show_matching_entry
    program_opts = dict()
    show_opts = dict(use_clipboard=is_clipboard_supported())
    # Parse the command line arguments.
    try:
        options, arguments = getopt.gnu_getopt(sys.argv[1:], 'elnp:vqh', [
            'edit', 'list', 'no-clipboard', 'password-store=',
            'verbose', 'quiet', 'help',
        ])
        for option, value in options:
            if option in ('-e', '--edit'):
                action = edit_matching_entry
            elif option in ('-l', '--list'):
                action = list_matching_entries
            elif option in ('-n', '--no-clipboard'):
                show_opts['use_clipboard'] = False
            elif option in ('-p', '--password-store'):
                stores = program_opts.setdefault('stores', [])
                stores.append(PasswordStore(directory=value))
            elif option in ('-v', '--verbose'):
                coloredlogs.increase_verbosity()
            elif option in ('-q', '--quiet'):
                coloredlogs.decrease_verbosity()
            elif option in ('-h', '--help'):
                usage(__doc__)
                return
            else:
                raise Exception("Unhandled option! (programming error)")
        if not (arguments or action == list_matching_entries):
            usage(__doc__)
            return
    except Exception as e:
        warning("Error: %s", e)
        sys.exit(1)
    # Execute the requested action.
    try:
        action(QuickPass(**program_opts), arguments,
               **(show_opts if action == show_matching_entry else {}))
    except PasswordStoreError as e:
        # Known issues don't get a traceback.
        logger.error("%s", e)
        sys.exit(1)
    except KeyboardInterrupt:
        # If the user interrupted an interactive prompt they most likely did so
        # intentionally, so there's no point in generating more output here.
        sys.exit(1)

def main(args=sys.argv):
    p = argparse.ArgumentParser(description='Print with the AxiDraw.')
    p.add_argument('--verbose', action='store_true')
    p.add_argument('--mock', action='store_true')
    p.set_defaults(function=None)

    subparsers = p.add_subparsers(help='sub-command help')

    p_plan = subparsers.add_parser(
        'plan', help='Plan the plotting actions for an SVG file.')
    p_plan.add_argument('filename')
    p_plan.add_argument('--out')
    p_plan.add_argument('--overwrite', action='store_true')
    p_plan.set_defaults(function=plan)

    p_plot = subparsers.add_parser(
        'plot', help='Plot an SVG file directly.')
    p_plot.add_argument('filename')
    p_plot.set_defaults(function=plot)

    p_info = subparsers.add_parser(
        'info', help='Print information about an SVG file.')
    p_info.add_argument('filename')
    p_info.set_defaults(function=info)

    p_server = subparsers.add_parser(
        'server', help='Run a server for remote plotting.')
    p_server.add_argument('--port', type=int, default=8888)
    p_server.set_defaults(function=server)

    p_manual = subparsers.add_parser(
        'manual', help='Manual control shell.')
    p_manual.add_argument('cmd', nargs='*')
    p_manual.set_defaults(function=manual)

    opts, args = p.parse_known_args(args[1:])

    if coloredlogs:
        coloredlogs.install(level='DEBUG' if opts.verbose else 'INFO')
    else:
        logging.basicConfig(level=logging.DEBUG if opts.verbose else
                            logging.INFO)

    if opts.function:
        return opts.function(opts)
    else:
        p.print_help()

def app_main(self):
        # Backup original arguments for later parsing
        args_src = sys.argv

        # Parse our argument list
        parser = argparse.ArgumentParser(description='Chat roulette python server')
        parser.add_argument('-n', '--non-interactive', dest='noninteractive', action='store_const', const=True,
                            help='non-interactive mode of operation, command line only')
        parser.add_argument('-r', '--attempts', dest='attempts', type=int, default=3,
                            help='number of attempts in non-interactive mode')
        parser.add_argument('-l','--pid-lock', dest='pidlock', type=int, default=-1,
                            help='number of attempts for pidlock acquire')
        parser.add_argument('--debug', dest='debug', action='store_const', const=True,
                            help='enables debug mode')
        parser.add_argument('--verbose', dest='verbose', action='store_const', const=True,
                            help='enables verbose mode')
        parser.add_argument('--force', dest='force', action='store_const', const=True, default=False,
                            help='forces some action')

        parser.add_argument('commands', nargs=argparse.ZERO_OR_MORE, default=[],
                            help='commands to process')

        self.args = parser.parse_args(args=args_src[1:])
        self.noninteractive = self.args.noninteractive

        # Fixing cmd2 arg parsing, call cmdLoop
        sys.argv = [args_src[0]]
        for cmd in self.args.commands:
            sys.argv.append(cmd)

        # Terminate after execution is over on the non-interactive mode
        if self.noninteractive:
            sys.argv.append('quit')

        if self.args.debug:
            coloredlogs.install(level=logging.DEBUG)

        self.cmdloop()
        sys.argv = args_src

        # Noninteractive - return the last result from the operation (for scripts)
        if self.noninteractive:
            sys.exit(self.last_result)

def retry(retries=CONFIG.CRAWLER.RETRY or 3, sleep=CONFIG.CRAWLER.SLEEP,
          changeip=False):
    """??????????????????????retrying
    pip install retrying
    https://github.com/rholder/retrying
    ??????????
    http://mp.weixin.qq.com/s?__biz=MzAwMDU1MTE1OQ==&mid=2653547274&idx=1&sn=52e5037b163146c1656eedce2da1ecd8&scene=1&srcid=0527MEXhNRZATtlTPhinD5Re#rd

    :param retries: number int of retry times.
    301 Moved Temporarily
    401 Unauthorized
    403 Forbidden
    404 Not Found
    408 Request Timeout
    429 Too Many Requests
    503 Service Unavailable
    """
    def _retry(func):
        @wraps(func)
        def _wrapper(*args, **kwargs):
            index = 0
            while index < retries:
                index += 1
                try:
                    response = func(*args, **kwargs)
                    if response.status_code in (301, 302, 404, 500):
                        print('status_code', response.status_code)
                        break
                    elif response.status_code != 200:
                        print(response.status_code)
                        if changeip:
                            change_ip()
                        continue
                    else:
                        break
                except Exception as e:
                    # ???????????
                    # traceback.print_exc()
                    response = None
                    if isinstance(e, Timeout):
                        if sleep is not None:
                            time.sleep(sleep + randint(1, 10))
                        continue
                    elif isinstance(e, TooManyRedirects):
                        break

            return response
        return _wrapper
    return _retry

def setup_logging(is_quiet: bool = False,
                  is_debug: bool = False,
                  verbosity: int = 6,
                  log_file: str = 'openadms.log') -> None:
    """Setups the logger and logging handlers.

    Args:
        is_quiet: Disable output.
        is_debug: Print debug messages.
        verbosity: Verbosity level (1 - 9).
        log_file: Path of the log file.
    """
    # Enable verbose logging.
    verboselogs.install()

    # Basic logging configuration.
    console_level = logging.DEBUG if is_debug else logging.INFO
    root.setLevel(console_level)

    fmt = '%(asctime)s - %(levelname)8s - %(name)26s - %(message)s'
    date_fmt = '%Y-%m-%dT%H:%M:%S'
    formatter = logging.Formatter(fmt, date_fmt)

    # File handler.
    file_level = {
        1: logging.CRITICAL,
        2: logging.ERROR,
        3: verboselogs.SUCCESS,
        4: logging.WARNING,
        5: verboselogs.NOTICE,
        6: logging.INFO,
        7: verboselogs.VERBOSE,
        8: logging.DEBUG,
        9: verboselogs.SPAM
    }.get(verbosity, 6)

    fh = logging.handlers.RotatingFileHandler(log_file,
                                              maxBytes=LOG_FILE_MAX_SIZE,
                                              backupCount=LOG_FILE_BACKUP_COUNT,
                                              encoding='utf8')
    fh.setLevel(file_level)
    fh.setFormatter(formatter)
    root.addHandler(fh)

    if is_quiet:
        # Silence logger output.
        console_level = 100

    # Colorized output of log messages.
    coloredlogs.install(level=console_level,
                        fmt=fmt,
                        datefmt=date_fmt,
                        logger=root)

    # Add filter to log handlers, to exclude log messages from HBMQTT.
    for handler in logging.root.handlers:
        handler.addFilter(RootFilter())