Python traceback 模块，print_tb() 实例源码

def mqtt_connect(mqtt_client, core_info):
    connected = False

    # try connecting to all connectivity info objects in the list
    for connectivity_info in core_info.connectivityInfoList:
        core_host = connectivity_info.host
        core_port = connectivity_info.port
        logging.info("Connecting to Core at {0}:{1}".format(
            core_host, core_port))
        mqtt_client.configureEndpoint(core_host, core_port)
        try:
            mqtt_client.connect()
            connected = True
            break
        except socket.error as se:
            print("SE:{0}".format(se))
        except operationTimeoutException as te:
            print("operationTimeoutException:{0}".format(te.message))
            traceback.print_tb(te, limit=25)
        except Exception as e:
            print("Exception caught:{0}".format(e.message))

    return connected

def mqtt_connect(mqtt_client, core_info):
    connected = False

    # try connecting to all connectivity info objects in the list
    for connectivity_info in core_info.connectivityInfoList:
        core_host = connectivity_info.host
        core_port = connectivity_info.port
        logging.info("Connecting to Core at {0}:{1}".format(
            core_host, core_port))
        mqtt_client.configureEndpoint(core_host, core_port)
        try:
            mqtt_client.connect()
            connected = True
            break
        except socket.error as se:
            print("SE:{0}".format(se))
        except operationTimeoutException as te:
            print("operationTimeoutException:{0}".format(te.message))
            traceback.print_tb(te, limit=25)
        except Exception as e:
            print("Exception caught:{0}".format(e.message))

    return connected

def handle_error(self, wrapper, exception, traceback_):
        print >> sys.stderr
        print >> sys.stderr, "---- location:"
        traceback.print_stack()
        print >> sys.stderr, "---- error:"
        traceback.print_tb(traceback_)
        try:
            stack = wrapper.stack_where_defined
        except AttributeError:
            print >> sys.stderr, "??:??: %s / %r" % (wrapper, exception)
        else:
            stack = list(stack)
            stack.reverse()
            for (filename, line_number, function_name, text) in stack:
                file_dir = os.path.dirname(os.path.abspath(filename))
                if file_dir.startswith(this_script_dir):
                    print >> sys.stderr, "%s:%i: %r" % (os.path.join("..", "bindings", "python", os.path.basename(filename)),
                                                        line_number, exception)
                    break
        return True

def on_command_error(error, ctx):
    if isinstance(error, commands.NoPrivateMessage):
        await bot.send_typing(ctx.message.author)
        await asyncio.sleep(1)
        await bot.send_message(ctx.message.author, "Um... this command can't be used in private messages.")
    elif isinstance(error, commands.DisabledCommand):
        await bot.send_typing(ctx.message.author)
        await asyncio.sleep(1)
        await bot.send_message(ctx.message.author, "I'm Sorry. This command is disabled and it can't be used.")
    elif isinstance(error, commands.CommandInvokeError):
        print('In {0.command.qualified_name}:'.format(ctx), file=sys.stderr)
        traceback.print_tb(error.original.__traceback__)
        print('{0.__class__.__name__}: {0}'.format(error.original), file=sys.stderr)
    elif isinstance(error, commands.CommandNotFound):
        log.info("'{0.message.author}' in {0.message.server} used a command thats not in Inkxbot, the content is resent here: '{0.message.content}'".format(ctx))
    elif isinstance(error, commands.MissingRequiredArgument):
        log.info("'{0.message.author}' in {0.message.server} was missing some arguments in a command, message is resent here: '{0.message.content}'".format(ctx))
        await bot.send_typing(ctx.message.channel)
        await asyncio.sleep(1)
        await bot.send_message(ctx.message.channel, "It seems you are missing required argument(s). Try again if you have all the arguments needed.")

def get_dispatch(distro, arg):
    """
    Dispatch to the appropriate scraping function for the given distro.
    """
    defin = DISTROS[distro]
    try:
        if isinstance(defin, type(lambda:None)):
            return defin(arg)
        elif isinstance(defin, type("")):
            return get_dispatch(defin, arg)
        else:
            return get_scrape(distro, defin, arg)
    except:
        ex_ty, ex_ob, ex_tb = sys.exc_info()
        trace('get_dispatch(%r, %r) failed:' % (distro, arg))
        if VERBOSE:
            import traceback
            traceback.print_tb(ex_tb)
        trace('  %s: %s' % (ex_ty.__name__, ex_ob))
        return parse_semver('0.0.0')

def boto_except_hook(debugger_flag, debug_flag):
    def excepthook(typ, value, tb):
        if typ is bdb.BdbQuit:
            sys.exit(1)
        sys.excepthook = sys.__excepthook__

        if debugger_flag and sys.stdout.isatty() and sys.stdin.isatty():
            if debugger.__name__ == 'epdb':
                debugger.post_mortem(tb, typ, value)
            else:
                debugger.post_mortem(tb)
        elif debug_flag:
            print(traceback.print_tb(tb))
            sys.exit(1)
        else:
            print(value)
            sys.exit(1)

    return excepthook

def on_command_error(ctx, error):
    if isinstance(error, commands.NoPrivateMessage):
        await ctx.send(content='This command cannot be used in private messages.')

    elif isinstance(error, commands.DisabledCommand):
        await ctx.send(content='This command is disabled and cannot be used.')

    elif isinstance(error, commands.MissingRequiredArgument):
        await bot.formatter.format_help_for(ctx, ctx.command, "You are missing required arguments.")

    elif isinstance(error, commands.CommandNotFound):
        # await ctx.send('Command not found. (I\'m working on fixing cmd_not_found forf help module')
        await bot.formatter.format_help_for(ctx, [c for c in bot.commands if 'help' == c.name][0],
                                            "Command not found.")

    elif isinstance(error, commands.CommandInvokeError):
        print('In {0.command.qualified_name}:'.format(ctx), file=sys.stderr)
        print('{0.__class__.__name__}: {0}'.format(error.original), file=sys.stderr)
        traceback.print_tb(error.__traceback__, file=sys.stderr)
        log.error('In {0.command.qualified_name}:'.format(ctx))
        log.error('{0.__class__.__name__}: {0}'.format(error.original))

    else:
        traceback.print_tb(error.__traceback__, file=sys.stderr)

def mock_input(testfunc):
    """Decorator for tests of the main interact loop.

    Write the test as a generator, yield-ing the input strings, which IPython
    will see as if they were typed in at the prompt.
    """
    def test_method(self):
        testgen = testfunc(self)
        with mock_input_helper(testgen) as mih:
            mih.ip.interact()

        if mih.exception is not None:
            # Re-raise captured exception
            etype, value, tb = mih.exception
            import traceback
            traceback.print_tb(tb, file=sys.stdout)
            del tb  # Avoid reference loop
            raise value

    return test_method

# Test classes -----------------------------------------------------------------

def load_wsgi(self):
        try:
            self.wsgi = self.app.wsgi()
        except SyntaxError as e:
            if self.cfg.reload == 'off':
                raise

            self.log.exception(e)

            # fix from PR #1228
            # storing the traceback into exc_tb will create a circular reference.
            # per https://docs.python.org/2/library/sys.html#sys.exc_info warning,
            # delete the traceback after use.
            try:
                exc_type, exc_val, exc_tb = sys.exc_info()
                self.reloader.add_extra_file(exc_val.filename)

                tb_string = six.StringIO()
                traceback.print_tb(exc_tb, file=tb_string)
                self.wsgi = util.make_fail_app(tb_string.getvalue())
            finally:
                del exc_tb

def _run_batches_from_queue(self):
        skip_batch = False
        while not self._stop_flag:
            future = self._prefetch_queue.get(block=True)
            if future is None:
                self._prefetch_queue.task_done()
                self._batch_queue.put(None)
                break
            else:
                try:
                    batch = future.result()
                except SkipBatchException:
                    skip_batch = True
                except Exception:   # pylint: disable=broad-except
                    exc = future.exception()
                    print("Exception in a thread:", exc)
                    traceback.print_tb(exc.__traceback__)
                finally:
                    if not skip_batch:
                        self._batch_queue.put(batch, block=True)
                        skip_batch = False
                    self._prefetch_queue.task_done()
        return None

def assemble(self, all_res, *args, **kwargs):
        """ Assemble the batch after a parallel action """
        _ = args, kwargs
        if any_action_failed(all_res):
            all_errors = self.get_errors(all_res)
            print(all_errors)
            traceback.print_tb(all_errors[0].__traceback__)
            raise RuntimeError("Could not assemble the batch")

        components = kwargs.get('components', 'images')
        if isinstance(components, (list, tuple)):
            all_res = list(zip(*all_res))
        else:
            components = [components]
            all_res = [all_res]
        for component, res in zip(components, all_res):
            self.assemble_component(res, component)
        return self

def assemble(self, all_res, *args, **kwargs):
        """ Assemble the batch after a parallel action """
        _ = args, kwargs
        if any_action_failed(all_res):
            all_errors = self.get_errors(all_res)
            print(all_errors)
            traceback.print_tb(all_errors[0].__traceback__)
            raise RuntimeError("Could not assemble the batch")

        components = kwargs.get('components', 'images')
        if isinstance(components, (list, tuple)):
            all_res = list(zip(*all_res))
        else:
            components = [components]
            all_res = [all_res]
        for component, res in zip(components, all_res):
            new_data = np.array(res, dtype='object')
            setattr(self, component, new_data)
        return self

def pyerr_write_unraisable(exc, obj):
    f = sys.stderr

    if obj:
        f.write("Exception ignored in: ")
        f.write(repr(obj))
        f.write("\n")

    t, v, tb = sys.exc_info()
    traceback.print_tb(tb, file=f)

    assert isinstance(v, Exception)
    f.write(t.__module__ + "." + t.__name__)
    f.write(": ")
    f.write(str(v))
    f.write("\n")

def step_impl(context):
    """

    :type context behave.runner.Context

    """

    _json_data = json_load_file(os.path.join(script_location, "../data/bad_json.json"))

    try:

        context.schema_tools.apply(_json_data)
    except Exception as e:
        if str(e)[:32] == "'canRead' is a required property":
            ok_(True)
        else:
            if hasattr(sys, "last_traceback"):
                _traceback = traceback.print_tb(sys.last_traceback)
            else:
                _traceback = "Not available"

            ok_(False, "Error: " + str(e) + "\nTraceback:" + _traceback)

def _in_thread_run_script(self):
        """ Run the script. """
        try:
            self.run_script()

        except Exception as e:
            # handle any exception thrown
            exc_type,exc_value,exc_traceback = sys.exc_info()
            print "Exception Error:  ",  exc_value
            print e

            # print traceback, excluding this file
            traceback.print_tb(exc_traceback)
            # exc_strings = traceback.format_list(traceback.extract_tb(exc_traceback))
            # exc_strings = [s for s in exc_strings if s.find("parlay_script.py")< 0 ]
            # for s in exc_strings:
            #     print s

def boto_except_hook(debugger_flag, debug_flag):
    def excepthook(typ, value, tb):
        if typ is bdb.BdbQuit:
            sys.exit(1)
        sys.excepthook = sys.__excepthook__

        if debugger_flag and sys.stdout.isatty() and sys.stdin.isatty():
            if debugger.__name__ == 'epdb':
                debugger.post_mortem(tb, typ, value)
            else:
                debugger.post_mortem(tb)
        elif debug_flag:
            print(traceback.print_tb(tb))
            sys.exit(1)
        else:
            print(value)
            sys.exit(1)

    return excepthook

def assertion_msg(message: str, *args):
    """This will:
        * print the custom assertion message
        * print the traceback (stack trace)
        * raise the original AssertionError exception

    :param message: a message that will be printed & logged when assertion fails
    :param args: values that will replace % conversion specifications in message
                 like: %s, %d
    """
    try:
        yield
    except AssertionError as e:
        if args:
            message = message % args
        logging.error(message)
        e.args += (message,)
        _, _, tb = sys.exc_info()
        traceback.print_tb(tb)
        raise

def assertion_msg(message: str, *args):
    """This will:
        * print the custom assertion message
        * print the traceback (stack trace)
        * raise the original AssertionError exception

    :param message: message that will be printed & logged when assertion fails
    :param args: values that will replace % conversion specifications in
                 message like: %s, %d
    """
    try:
        yield
    except AssertionError as e:
        if args:
            message = message % args
        logging.error(message)
        e.args += (message,)
        _, _, tb = sys.exc_info()
        traceback.print_tb(tb)
        raise

def run(self):
        try:
            open_luggage = luggage.Luggage(luggage_path=self.path, password=self.password,
                                           pyluggage_config=self.pyluggage_config)
            self.open_luggage = open_luggage
        except Exception as ex:
            if be_verbose:
                print ">>>> Error opening Luggage"
                import sys
                import traceback
                exc_type, exc_value, exc_traceback = sys.exc_info()
                print "*** print_tb:"
                traceback.print_tb(exc_traceback, limit=1, file=sys.stdout)
                print "*** print_exception:"
                traceback.print_exception(exc_type, exc_value, exc_traceback,
                                          limit=2, file=sys.stdout)
            self.error = ex
        finally:
            self.finished.emit()

def run(self):
        try:
            self.new_luggage = luggage.Luggage.create_new_luggage(
                luggage_path=self.path,
                password=self.password,
                pyluggage_config=self.pyluggage_config)
        except Exception as ex:
            if be_verbose:
                print ">>>> Error creating Luggage"
                import sys
                import traceback
                exc_type, exc_value, exc_traceback = sys.exc_info()
                print "*** print_tb:"
                traceback.print_tb(exc_traceback, limit=1, file=sys.stdout)
                print "*** print_exception:"
                traceback.print_exception(exc_type, exc_value, exc_traceback, file=sys.stdout)
            self.error = ex
        finally:
            self.finished.emit()

def play_music(self, ctx, *, song : str):
    state = self.get_voice_state(ctx.message.server)
    opts = {
      "default_search": "auto",
      "quiet": True
    }

    if state.voice is None:
      success = await ctx.invoke(self.summon)
      if not success:
        return
    try:
      player =  await state.voice.create_ytdl_player(song, ytdl_options=opts, after=state.toggle_next)
    except Exception as e:
      traceback.print_tb(e.__traceback__)
      await self.bot.send_message(ctx.message.channel, "Could not play song: [{}]: {}".format(type(e).__name__, e))
    else:
      player.volume = 0.5
      player_state = VoicePlayer(ctx.message, player, state.voice)
      await self.bot.say("Queued **{}**".format(player.title))
      await state.songs.put(player_state)
      state.song_list.append(player_state)

def on_command_error(self, error, ctx):
        if isinstance(error, commands.NoPrivateMessage):
            await self.send_message(ctx.message.author, "\N{WARNING SIGN} Sorry, you can't use this command in a private message!")

        elif isinstance(error, commands.DisabledCommand):
            await self.send_message(ctx.message.author, "\N{WARNING SIGN} Sorry, this command is disabled!")

        elif isinstance(error, commands.CommandOnCooldown):
            await self.send_message(ctx.message.channel, f"{ctx.message.author.mention} slow down! Try again in {error.retry_after:.1f} seconds.")

        elif isinstance(error, commands.MissingRequiredArgument) or isinstance(error, commands.BadArgument):
            await self.send_message(ctx.message.channel, f"\N{WARNING SIGN} {error}")

        elif isinstance(error, commands.CommandInvokeError):
            original_name = error.original.__class__.__name__
            print(f"In {paint(ctx.command.qualified_name, 'b_red')}:")
            traceback.print_tb(error.original.__traceback__)
            print(f"{paint(original_name, 'red')}: {error.original}")

        else:
            print(f"{paint(type(error).__name__, 'b_red')}: {error}")

def handle_indexing(app, mysql):
    with app.app_context():
        while(True):
            try:
                g.conn = mysql.connect()
                g.cursor = g.conn.cursor()
                g.conn.begin()
                # run indexing servie every 300 seconds
                time.sleep(300)
                sqlpie.Indexer().index_documents()
                g.conn.commit()
            except Exception as e:
                if sqlpie.Util.is_debug():
                    traceback.print_tb(sys.exc_info()[2])
                try:
                    g.conn.rollback()
                except:
                    pass
            finally:
                #  if the MySQL Server is not running, this will fail.
                try:
                    g.cursor.close()
                    g.conn.close()
                except:
                    pass

def get(data, pattern, typ, windex, st=0):
  x = None
  for line in data.lines:
    if pattern in line:
      try:
        x = line.split()[windex]
        if st < 0:
          x = x[0:st]
        else:
          x = x[st:]
        x = typ(x)
      except:
        print line.split()
        print x, st
        print sys.exc_info()[1]
        traceback.print_tb(sys.exc_info()[2])
        quit()
      return x
  return x

def test_push_dirty_git(runner, monkeypatch):
    with runner.isolated_filesystem():
        os.mkdir('wandb')

        # If the test was run from a directory containing .wandb, then __stage_dir__
        # was '.wandb' when imported by api.py, reload to fix. UGH!
        reload(wandb)
        repo = git_repo()
        open("foo.txt", "wb").close()
        repo.repo.index.add(["foo.txt"])
        monkeypatch.setattr(cli, 'api', wandb_api.Api({'project': 'test'}))
        cli.api._settings['git_tag'] = True
        result = runner.invoke(
            cli.push, ["test", "foo.txt", "-p", "test", "-m", "Dirty"])
        print(result.output)
        print(result.exception)
        print(traceback.print_tb(result.exc_info[2]))
        assert result.exit_code == 1
        assert "You have un-committed changes." in result.output

def test_push_dirty_force_git(runner, request_mocker, query_project, upload_url, upsert_run, monkeypatch):
    query_project(request_mocker)
    upload_url(request_mocker)
    update_mock = upsert_run(request_mocker)
    with runner.isolated_filesystem():
        # So GitRepo is in this cwd
        monkeypatch.setattr(cli, 'api', wandb_api.Api({'project': 'test'}))
        repo = git_repo()
        with open('weights.h5', 'wb') as f:
            f.write(os.urandom(100))
        repo.repo.index.add(["weights.h5"])
        result = runner.invoke(
            cli.push, ["test", "weights.h5", "-f", "-p", "test", "-m", "Dirty"])
        print(result.output)
        print(result.exception)
        print(traceback.print_tb(result.exc_info[2]))
        assert result.exit_code == 0

def test_restore(runner, request_mocker, query_run, monkeypatch):
    mock = query_run(request_mocker)
    with runner.isolated_filesystem():
        os.mkdir("wandb")
        repo = git_repo()
        with open("patch.txt", "w") as f:
            f.write("test")
        repo.repo.index.add(["patch.txt"])
        repo.repo.commit()
        monkeypatch.setattr(cli, 'api', wandb_api.Api({'project': 'test'}))
        result = runner.invoke(cli.restore, ["test/abcdef"])
        print(result.output)
        print(traceback.print_tb(result.exc_info[2]))
        assert result.exit_code == 0
        assert "Created branch wandb/abcdef" in result.output
        assert "Applied patch" in result.output
        assert "Restored config variables" in result.output

def test_init_new_login(runner, empty_netrc, local_netrc, request_mocker, query_projects, query_viewer):
    query_viewer(request_mocker)
    query_projects(request_mocker)
    with runner.isolated_filesystem():
        # If the test was run from a directory containing .wandb, then __stage_dir__
        # was '.wandb' when imported by api.py, reload to fix. UGH!
        reload(wandb)
        result = runner.invoke(cli.init, input="%s\nvanpelt" % DUMMY_API_KEY)
        print('Output: ', result.output)
        print('Exception: ', result.exception)
        print('Traceback: ', traceback.print_tb(result.exc_info[2]))
        assert result.exit_code == 0
        with open("netrc", "r") as f:
            generatedNetrc = f.read()
        with open("wandb/settings", "r") as f:
            generatedWandb = f.read()
        assert DUMMY_API_KEY in generatedNetrc
        assert "test_model" in generatedWandb

def test_init_add_login(runner, empty_netrc, local_netrc, request_mocker, query_projects, query_viewer):
    query_viewer(request_mocker)
    query_projects(request_mocker)
    with runner.isolated_filesystem():
        with open("netrc", "w") as f:
            f.write("previous config")
        result = runner.invoke(cli.init, input="%s\nvanpelt\n" % DUMMY_API_KEY)
        print(result.output)
        print(result.exception)
        print(traceback.print_tb(result.exc_info[2]))
        assert result.exit_code == 0
        with open("netrc", "r") as f:
            generatedNetrc = f.read()
        with open("wandb/settings", "r") as f:
            generatedWandb = f.read()
        assert DUMMY_API_KEY in generatedNetrc
        assert "previous config" in generatedNetrc

def mock_input(testfunc):
    """Decorator for tests of the main interact loop.

    Write the test as a generator, yield-ing the input strings, which IPython
    will see as if they were typed in at the prompt.
    """
    def test_method(self):
        testgen = testfunc(self)
        with mock_input_helper(testgen) as mih:
            mih.ip.interact()

        if mih.exception is not None:
            # Re-raise captured exception
            etype, value, tb = mih.exception
            import traceback
            traceback.print_tb(tb, file=sys.stdout)
            del tb  # Avoid reference loop
            raise value

    return test_method

# Test classes -----------------------------------------------------------------

def boto_except_hook(debugger_flag, debug_flag):
    def excepthook(typ, value, tb):
        if typ is bdb.BdbQuit:
            sys.exit(1)
        sys.excepthook = sys.__excepthook__

        if debugger_flag and sys.stdout.isatty() and sys.stdin.isatty():
            if debugger.__name__ == 'epdb':
                debugger.post_mortem(tb, typ, value)
            else:
                debugger.post_mortem(tb)
        elif debug_flag:
            print traceback.print_tb(tb)
            sys.exit(1)
        else:
            print value
            sys.exit(1)

    return excepthook

def createDrawingCoords(self):
        if self.drawingCoords is None:
            self.drawingCoords = []
            self.pathIterator = self._tokenize_path(self.d)
            while True:
                try:
                    command = next(self.pathIterator)
                    if not command in self.COMMANDS:
                        self.additionalParameter = float(command)
                        coord = self.svgAdditionalCoords.get(self.currentCommand)()
                    else:
                        coord = self.svgPathSegmentCommands.get(command.strip(), self.unknownCommand)()
                        self.currentCommand = command.strip()
                    if coord is not None:
                        self.drawingCoords.append(coord)
                except StopIteration:
                    break
                except:
                    exc_type, exc_value, exc_traceback = sys.exc_info()
                    print("test1 main thread exception : %s" % exc_type)
                    traceback.print_tb(exc_traceback, limit=2, file=sys.stdout)
        return self.drawingCoords

def createDrawingCoords(self):
        if self.drawingCoords is None:
            self.drawingCoords = []
            tokens = self._tokenize_path(self.points)
            x = float(next(tokens))
            y = float(next(tokens))
            self.drawingCoords.append(Coordinate.fromCoords(x, y, True))
            while True:
                try:
                    self.drawingCoords.append(Coordinate.fromCoords(float(next(tokens)), float(next(tokens)), False))
                except StopIteration:
                    break
                except:
                    exc_type, exc_value, exc_traceback = sys.exc_info()
                    print("test1 main thread exception : %s" % exc_type)
                    traceback.print_tb(exc_traceback, limit=2, file=sys.stdout)
        return self.drawingCoords

def createDrawingCoords(self):
        if self.drawingCoords is None:
            self.drawingCoords = []
            tokens = self._tokenize_path(self.points)
            x = float(next(tokens))
            y = float(next(tokens))
            startX, startY = x, y
            self.drawingCoords.append(Coordinate.fromCoords(x, y, True))
            while True:
                try:
                    self.drawingCoords.append(Coordinate.fromCoords(float(next(tokens)), float(next(tokens)), False))
                except StopIteration:
                    break
                except:
                    exc_type, exc_value, exc_traceback = sys.exc_info()
                    print("test1 main thread exception : %s" % exc_type)
                    traceback.print_tb(exc_traceback, limit=2, file=sys.stdout)
            self.drawingCoords.append(Coordinate.fromCoords(startX, startY, False))

        return self.drawingCoords

def __exit__(self, type, value, traceback):

        """ stop the running app and clean up """

        conf = self.get_config()

        playbook = os.path.join(directories.workspace, conf["teardown"]["ansible"])

        self.subprocess.call([
            "ansible-playbook", playbook
        ])

        if traceback:
            logging.error("PR Build failed")
            logging.error(str(traceback))
            logging.error(str(value))
            tb.print_tb(traceback)

def boto_except_hook(debugger_flag, debug_flag):
    def excepthook(typ, value, tb):
        if typ is bdb.BdbQuit:
            sys.exit(1)
        sys.excepthook = sys.__excepthook__

        if debugger_flag and sys.stdout.isatty() and sys.stdin.isatty():
            if debugger.__name__ == 'epdb':
                debugger.post_mortem(tb, typ, value)
            else:
                debugger.post_mortem(tb)
        elif debug_flag:
            print traceback.print_tb(tb)
            sys.exit(1)
        else:
            print value
            sys.exit(1)

    return excepthook

def die(message, excode=1, exception=None):
    """
    Print error message into srdErr
    :param message: message
    :param excode: exitcode
    :param exception: exception for debugging mode
    """
    global PROGRESS
    if PROGRESS is not None:
        if sys.stderr.isatty():
            progress(result="\x1b[31;01mFAIL\x1b[39;49;00m")
        else:
            progress(result="FAIL")
    global DEBUG
    if exception and DEBUG:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        sys.stderr.write("\n\x1b[92;01m")
        traceback.print_tb(exc_traceback)
        sys.stderr.write("\x1b[39;49;00m\n")
    if sys.stderr.isatty():
        sys.stderr.write("\n\x1b[31;01m%s\x1b[39;49;00m\n\n" % message)
    else:
        sys.stderr.write("\n%s\n\n" % message)
    sys.exit(excode)

def runEngine(self, context):
        """Selects a random Move object based on its
        assigned probability and and performs its move() function
        on a context.
        Parameters
        ----------
        context : openmm.context object
        OpenMM context whose positions should be moved.
        """
        try:
            new_context = self.moves[self.selected_move].move(context)
        except Exception as e:
            #In case the move isn't properly implemented, print out useful info
            print('Error: move not implemented correctly, printing traceback:')
            ex_type, ex, tb = sys.exc_info()
            traceback.print_tb(tb)
            print(e)
            raise SystemExit

        return new_context

def boto_except_hook(debugger_flag, debug_flag):
    def excepthook(typ, value, tb):
        if typ is bdb.BdbQuit:
            sys.exit(1)
        sys.excepthook = sys.__excepthook__

        if debugger_flag and sys.stdout.isatty() and sys.stdin.isatty():
            if debugger.__name__ == 'epdb':
                debugger.post_mortem(tb, typ, value)
            else:
                debugger.post_mortem(tb)
        elif debug_flag:
            print(traceback.print_tb(tb))
            sys.exit(1)
        else:
            print(value)
            sys.exit(1)

    return excepthook

def load_wsgi(self):
        try:
            self.wsgi = self.app.wsgi()
        except SyntaxError as e:
            if self.cfg.reload == 'off':
                raise

            self.log.exception(e)

            # fix from PR #1228
            # storing the traceback into exc_tb will create a circular reference.
            # per https://docs.python.org/2/library/sys.html#sys.exc_info warning,
            # delete the traceback after use.
            try:
                exc_type, exc_val, exc_tb = sys.exc_info()
                self.reloader.add_extra_file(exc_val.filename)

                tb_string = six.StringIO()
                traceback.print_tb(exc_tb, file=tb_string)
                self.wsgi = util.make_fail_app(tb_string.getvalue())
            finally:
                del exc_tb

def initLog(log_path):
    '''
    ??????????????
    '''
    try:
        if os.path.exists(log_path)==False:
            os.makedirs(log_path)
    except:
        etype, evalue, tracebackObj = sys.exc_info()[:3]
        print("Type: " , etype)
        print("Value: " , evalue)
        traceback.print_tb(tracebackObj)

    track_log = log_path+'/track.log'    

    logging.basicConfig(level=logging.DEBUG,
                format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
                datefmt='%a, %d %b %Y %H:%M:%S',
                filename=track_log,
                filemode='a')

def execute_command(self, command, slack_id, params, channel):
        """Attempt to run the command with the parameters provided."""
        self.acknowledge_command(channel)
        synapse_user = self.synapse_user_from_slack_user_id(slack_id)
        response = self.registration_prompt(slack_id)

        if synapse_user:
            try:
                response = command(slack_id, synapse_user, params)
            except SynapsePayError as e:
                response = (
                    'An HTTP error occurred while trying to communicate with '
                    'the Synapse API:\n{0}'.format(e.message)
                )
            except Exception as e:
                traceback.print_tb(e.__traceback__)
                response = 'An error occurred:\n{0}: {1}'.format(
                    sys.exc_info()[0],sys.exc_info()[1]
                )
        return response

def boto_except_hook(debugger_flag, debug_flag):
    def excepthook(typ, value, tb):
        if typ is bdb.BdbQuit:
            sys.exit(1)
        sys.excepthook = sys.__excepthook__

        if debugger_flag and sys.stdout.isatty() and sys.stdin.isatty():
            if debugger.__name__ == 'epdb':
                debugger.post_mortem(tb, typ, value)
            else:
                debugger.post_mortem(tb)
        elif debug_flag:
            print traceback.print_tb(tb)
            sys.exit(1)
        else:
            print value
            sys.exit(1)

    return excepthook

def mock_input(testfunc):
    """Decorator for tests of the main interact loop.

    Write the test as a generator, yield-ing the input strings, which IPython
    will see as if they were typed in at the prompt.
    """
    def test_method(self):
        testgen = testfunc(self)
        with mock_input_helper(testgen) as mih:
            mih.ip.interact()

        if mih.exception is not None:
            # Re-raise captured exception
            etype, value, tb = mih.exception
            import traceback
            traceback.print_tb(tb, file=sys.stdout)
            del tb  # Avoid reference loop
            raise value

    return test_method

# Test classes -----------------------------------------------------------------

def scrap(self):
        """
        scrap target url page
        """

        # get list of detail article url
        article_detail_url_list = self.get_article_detail_urls()

        article_detail_info = []
        for article_url in article_detail_url_list:
            try:
                article_dict = self.get_article_detail_info_dict(article_url)
                article_detail_info.append(article_dict)

            except AttributeError as err:
                print("[ Exception ] Exception occured in GScraper#scrap: {}".format(err))
                traceback.print_tb(err.__traceback__)

        self.save_article_detail_info_list_to_csv(article_detail_info)

def mock_input(testfunc):
    """Decorator for tests of the main interact loop.

    Write the test as a generator, yield-ing the input strings, which IPython
    will see as if they were typed in at the prompt.
    """
    def test_method(self):
        testgen = testfunc(self)
        with mock_input_helper(testgen) as mih:
            mih.ip.interact()

        if mih.exception is not None:
            # Re-raise captured exception
            etype, value, tb = mih.exception
            import traceback
            traceback.print_tb(tb, file=sys.stdout)
            del tb  # Avoid reference loop
            raise value

    return test_method

# Test classes -----------------------------------------------------------------

def _connect_once(self):
        try:
            yield from self._connect()

        except ConnectionRefusedError:
            self.print_debug("connection refused ({})".format(self.address))

        except OSError:
            self.print_debug("connection failed ({})".format(self.address))

        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            traceback.print_tb(exc_traceback, limit=1, file=sys.stdout)
            traceback.print_exception(exc_type, exc_value, exc_traceback,
                    file=sys.stdout)
            self.print_debug("connection failed ({})".format(self.address))

def onMessage(self, payload, isBinary):
        if isBinary:
            try:
                self.binaryHandler(payload)
            except KeyboardInterrupt:
                raise KeyboardInterrupt()

            except:
                exc_type, exc_value, exc_traceback = sys.exc_info()
                traceback.print_tb(exc_traceback, limit=1, file=sys.stdout)
                traceback.print_exception(exc_type, exc_value, exc_traceback,
                        limit=6, file=sys.stdout)
        else:
            try:
                if self.textHandler:
                    self.textHandler(payload)
            except KeyboardInterrupt:
                raise KeyboardInterrupt()
            except:
                exc_type, exc_value, exc_traceback = sys.exc_info()
                traceback.print_tb(exc_traceback, limit=1, file=sys.stdout)
                traceback.print_exception(exc_type, exc_value, exc_traceback,
                        limit=6, file=sys.stdout)

def broadcast(self, msg, isBinary = False):
        try:
            if self.throttle:
                self.broadcastMsg = msg
            else:
                if self.encodeMsg:
                    msg = base64.b64encode(self.msg)
                for c in self.clients:
                    if isBinary:
                        c.sendBinaryMsg(msg)
                    else:
                        c.sendTextMsg(msg)
        except:
            self.print_debug("exception while broadcast()")
            exc_type, exc_value, exc_traceback = sys.exc_info()
            traceback.print_tb(exc_traceback, limit=1, file=sys.stdout)
            traceback.print_exception(exc_type, exc_value, exc_traceback,
                          limit=6, file=sys.stdout)

def get_info(msg, info_type):
    global is_match
    get = {
        InputType.TRACK: _get_track,
        InputType.ALBUM: _get_album,
        InputType.PLAYLIST: _get_playlist,
        InputType.ARTIST: _get_artist,
    }
    is_match = False

    try:
        info = get[info_type](msg)
    except:
        tb = sys.exc_info()
        print(tb[1])
        print(traceback.print_tb(tb[2]))
        return {'mode':ErrorType.SOMETHING_WRONG}
    return info