ngrok_server.py 文件源码

def _read_ngrok_tunnels(self):
        webhook_urls = dict()
        self._api_url = 'http://%s:%d/api' % self._find_ngrok_inspection_port()
        logging.info('Local ngrok API url: %s', self._api_url)
        for _ in range(0, self.POLL_TUNNELS_MAX_TRIES):
            try:
                data = requests.get(self._api_url + '/tunnels').json()
                if 'tunnels' not in data or len(data['tunnels']) == 0:
                    raise ValueError('ngrok API did not return any tunnel.')
                for tunnel in data['tunnels']:
                    if tunnel['config']['addr'].endswith(':' + str(self.server.server_port)):
                        webhook_urls[tunnel['proto']] = tunnel['public_url']
                break
            except (requests.ConnectionError, ValueError) as e:
                logging.error('Error reading ngrok API: %s. Retry in 1sec.', e)
                time.sleep(1)
        if 'https' in webhook_urls:
            self._webhook_url = webhook_urls['https'] + '/' + self.server.session_token
        else:
            raise RuntimeError('Did not receive any HTTPS tunnel from ngrok API.')