import io
import signal
import asyncio
import warnings
from pprint import pprint
from functools import partial
from collections import defaultdict, Counter
from .errors import *
from .proxy import Proxy
from .server import Server
from .checker import Checker
from .utils import log, IPPortPatternLine
from .resolver import Resolver
from .providers import Provider, PROVIDERS
# Pause between grabbing cycles; in seconds.
GRAB_PAUSE = 180
# The maximum number of providers that are parsed concurrently
MAX_CONCURRENT_PROVIDERS = 3
[docs]class Broker:
"""The Broker.
| One broker to rule them all, one broker to find them,
| One broker to bring them all and in the darkness bind them.
:param asyncio.Queue queue: (optional) Queue of found/checked proxies
:param int timeout: (optional) Timeout of a request in seconds
:param int max_conn:
(optional) The maximum number of concurrent checks of proxies
:param int max_tries:
(optional) The maximum number of attempts to check a proxy
:param list judges:
(optional) Urls of pages that show HTTP headers and IP address.
Or :class:`~proxybroker.judge.Judge` objects
:param list providers:
(optional) Urls of pages where to find proxies.
Or :class:`~proxybroker.providers.Provider` objects
:param bool verify_ssl:
(optional) Flag indicating whether to check the SSL certificates.
Set to True to check ssl certifications
:param loop: (optional) asyncio compatible event loop
.. deprecated:: 0.2.0
Use :attr:`max_conn` and :attr:`max_tries` instead of
:attr:`max_concurrent_conn` and :attr:`attempts_conn`.
"""
def __init__(self, queue=None, timeout=8, max_conn=200, max_tries=3,
judges=None, providers=None, verify_ssl=False, loop=None,
**kwargs):
self._loop = loop or asyncio.get_event_loop()
self._proxies = queue or asyncio.Queue(loop=self._loop)
self._resolver = Resolver(loop=self._loop)
self._timeout = timeout
self._verify_ssl = verify_ssl
self.unique_proxies = {}
self._all_tasks = []
self._checker = None
self._server = None
self._limit = 0 # not limited
self._countries = None
max_concurrent_conn = kwargs.get('max_concurrent_conn')
if max_concurrent_conn:
warnings.warn(
'`max_concurrent_conn` is deprecated, use `max_conn` instead',
DeprecationWarning)
if isinstance(max_concurrent_conn, asyncio.Semaphore):
max_conn = max_concurrent_conn._value
else:
max_conn = max_concurrent_conn
attempts_conn = kwargs.get('attempts_conn')
if attempts_conn:
warnings.warn(
'`attempts_conn` is deprecated, use `max_tries` instead',
DeprecationWarning)
max_tries = attempts_conn
# The maximum number of concurrent checking proxies
self._on_check = asyncio.Queue(maxsize=max_conn, loop=self._loop)
self._max_tries = max_tries
self._judges = judges
self._providers = [p if isinstance(p, Provider) else Provider(p)
for p in (providers or PROVIDERS)]
try:
self._loop.add_signal_handler(signal.SIGINT, self.stop)
# add_signal_handler() is not implemented on Win
# https://docs.python.org/3.5/library/asyncio-eventloops.html#windows
except NotImplementedError:
pass
[docs] async def grab(self, *, countries=None, limit=0):
"""Gather proxies from the providers without checking.
:param list countries: (optional) List of ISO country codes
where should be located proxies
:param int limit: (optional) The maximum number of proxies
:ref:`Example of usage <proxybroker-examples-grab>`.
"""
self._countries = countries
self._limit = limit
task = asyncio.ensure_future(self._grab(check=False))
self._all_tasks.append(task)
[docs] async def find(self, *, types=None, data=None, countries=None,
post=False, strict=False, dnsbl=None, limit=0, **kwargs):
"""Gather and check proxies from providers or from a passed data.
:ref:`Example of usage <proxybroker-examples-find>`.
:param list types:
Types (protocols) that need to be check on support by proxy.
Supported: HTTP, HTTPS, SOCKS4, SOCKS5, CONNECT:80, CONNECT:25
And levels of anonymity (HTTP only): Transparent, Anonymous, High
:param data:
(optional) String or list with proxies. Also can be a file-like
object supports `read()` method. Used instead of providers
:param list countries:
(optional) List of ISO country codes where should be located proxies
:param bool post:
(optional) Flag indicating use POST instead of GET for requests
when checking proxies
:param bool strict:
(optional) Flag indicating that anonymity levels of types
(protocols) supported by a proxy must be equal to the requested
types and levels of anonymity. By default, strict mode is off and
for a successful check is enough to satisfy any one of the
requested types
:param list dnsbl:
(optional) Spam databases for proxy checking.
`Wiki <https://en.wikipedia.org/wiki/DNSBL>`_
:param int limit: (optional) The maximum number of proxies
:raises ValueError:
If :attr:`types` not given.
.. versionchanged:: 0.2.0
Added: :attr:`post`, :attr:`strict`, :attr:`dnsbl`.
Changed: :attr:`types` is required.
"""
ip = await self._resolver.get_real_ext_ip()
types = _update_types(types)
if not types:
raise ValueError('`types` is required')
self._checker = Checker(
judges=self._judges, timeout=self._timeout,
verify_ssl=self._verify_ssl, max_tries=self._max_tries,
real_ext_ip=ip, types=types, post=post,
strict=strict, dnsbl=dnsbl, loop=self._loop)
self._countries = countries
self._limit = limit
tasks = [asyncio.ensure_future(self._checker.check_judges())]
if data:
task = asyncio.ensure_future(self._load(data, check=True))
else:
task = asyncio.ensure_future(self._grab(types, check=True))
tasks.append(task)
self._all_tasks.extend(tasks)
[docs] def serve(self, host='127.0.0.1', port=8888, limit=100, **kwargs):
"""Start a local proxy server.
The server distributes incoming requests to a pool of found proxies.
When the server receives an incoming request, it chooses the optimal
proxy (based on the percentage of errors and average response time)
and passes to it the incoming request.
In addition to the parameters listed below are also accept all the
parameters of the :meth:`.find` method and passed it to gather proxies
to a pool.
:ref:`Example of usage <proxybroker-examples-server>`.
:param str host: (optional) Host of local proxy server
:param int port: (optional) Port of local proxy server
:param int limit:
(optional) When will be found a requested number of working
proxies, checking of new proxies will be lazily paused.
Checking will be resumed if all the found proxies will be discarded
in the process of working with them (see :attr:`max_error_rate`,
:attr:`max_resp_time`). And will continue until it finds one working
proxy and paused again. The default value is 100
:param int max_tries:
(optional) The maximum number of attempts to handle an incoming
request. If not specified, it will use the value specified during
the creation of the :class:`Broker` object. Attempts can be made
with different proxies. The default value is 3
:param int min_req_proxy:
(optional) The minimum number of processed requests to estimate the
quality of proxy (in accordance with :attr:`max_error_rate` and
:attr:`max_resp_time`). The default value is 5
:param int max_error_rate:
(optional) The maximum percentage of requests that ended with
an error. For example: 0.5 = 50%. If proxy.error_rate exceeds this
value, proxy will be removed from the pool.
The default value is 0.5
:param int max_resp_time:
(optional) The maximum response time in seconds.
If proxy.avg_resp_time exceeds this value, proxy will be removed
from the pool. The default value is 8
:param bool prefer_connect:
(optional) Flag that indicates whether to use the CONNECT method
if possible. For example: If is set to True and a proxy supports
HTTP proto (GET or POST requests) and CONNECT method, the server
will try to use CONNECT method and only after that send the
original request. The default value is False
:param list http_allowed_codes:
(optional) Acceptable HTTP codes returned by proxy on requests.
If a proxy return code, not included in this list, it will be
considered as a proxy error, not a wrong/unavailable address.
For example, if a proxy will return a ``404 Not Found`` response -
this will be considered as an error of a proxy.
Checks only for HTTP protocol, HTTPS not supported at the moment.
By default the list is empty and the response code is not verified
:param int backlog:
(optional) The maximum number of queued connections passed to
listen. The default value is 100
:raises ValueError:
If :attr:`limit` is less than or equal to zero.
Because a parsing of providers will be endless
.. versionadded:: 0.2.0
"""
if limit <= 0:
raise ValueError(
'In serve mode value of the limit cannot be less than or '
'equal to zero. Otherwise, a parsing of providers will be '
'endless')
self._server = Server(
host=host, port=port, proxies=self._proxies, timeout=self._timeout,
max_tries=kwargs.pop('max_tries', self._max_tries),
loop=self._loop, **kwargs)
self._server.start()
task = asyncio.ensure_future(self.find(limit=limit, **kwargs))
self._all_tasks.append(task)
async def _load(self, data, check=True):
"""Looking for proxies in the passed data.
Transform the passed data from [raw string | file-like object | list]
to set {(host, port), ...}: {('192.168.0.1', '80'), }
"""
log.debug('Load proxies from the raw data')
if isinstance(data, io.TextIOWrapper):
data = data.read()
if isinstance(data, str):
data = IPPortPatternLine.findall(data)
proxies = set(data)
for proxy in proxies:
await self._handle(proxy, check=check)
await self._on_check.join()
self._done()
async def _grab(self, types=None, check=False):
def _get_tasks(by=MAX_CONCURRENT_PROVIDERS):
providers = [pr for pr in self._providers if not types or
not pr.proto or bool(pr.proto & types.keys())]
while providers:
tasks = [asyncio.ensure_future(pr.get_proxies())
for pr in providers[:by]]
del providers[:by]
self._all_tasks.extend(tasks)
yield tasks
log.debug('Start grabbing proxies')
while True:
for tasks in _get_tasks():
for task in asyncio.as_completed(tasks):
proxies = await task
for proxy in proxies:
await self._handle(proxy, check=check)
log.debug('Grab cycle is complete')
if self._server:
log.debug('fall asleep for %d seconds' % GRAB_PAUSE)
await asyncio.sleep(GRAB_PAUSE)
log.debug('awaked')
else:
break
await self._on_check.join()
self._done()
async def _handle(self, proxy, check=False):
try:
proxy = await Proxy.create(
*proxy, timeout=self._timeout, resolver=self._resolver,
verify_ssl=self._verify_ssl, loop=self._loop)
except (ResolveError, ValueError):
return
if not self._is_unique(proxy) or not self._geo_passed(proxy):
return
if check:
await self._push_to_check(proxy)
else:
self._push_to_result(proxy)
def _is_unique(self, proxy):
if (proxy.host, proxy.port) not in self.unique_proxies:
self.unique_proxies[(proxy.host, proxy.port)] = proxy
return True
else:
return False
def _geo_passed(self, proxy):
if self._countries and (proxy.geo.code not in self._countries):
proxy.log('Location of proxy is outside the given countries list')
return False
else:
return True
async def _push_to_check(self, proxy):
def _task_done(proxy, f):
self._on_check.task_done()
if not self._on_check.empty():
self._on_check.get_nowait()
try:
if f.result():
# proxy is working and its types is equal to the requested
self._push_to_result(proxy)
except asyncio.CancelledError:
pass
if self._server and not self._proxies.empty() and self._limit <= 0:
log.debug('pause. proxies: %s; limit: %s' % (
self._proxies.qsize(), self._limit))
await self._proxies.join()
log.debug('unpause. proxies: %s' % self._proxies.qsize())
await self._on_check.put(None)
task = asyncio.ensure_future(self._checker.check(proxy))
task.add_done_callback(partial(_task_done, proxy))
self._all_tasks.append(task)
def _push_to_result(self, proxy):
log.debug('push to result: %r' % proxy)
self._proxies.put_nowait(proxy)
self._update_limit()
def _update_limit(self):
self._limit -= 1
if self._limit == 0 and not self._server:
self._done()
[docs] def stop(self):
"""Stop all tasks, and the local proxy server if it's running."""
self._done()
if self._server:
self._server.stop()
self._server = None
log.info('Stop!')
def _done(self):
log.debug('called done')
while self._all_tasks:
task = self._all_tasks.pop()
if not task.done():
task.cancel()
self._push_to_result(None)
log.info('Done! Total found proxies: %d' % len(self.unique_proxies))
[docs] def show_stats(self, verbose=False, **kwargs):
"""Show statistics on the found proxies.
Useful for debugging, but you can also use if you're interested.
:param verbose: Flag indicating whether to print verbose stats
.. deprecated:: 0.2.0
Use :attr:`verbose` instead of :attr:`full`.
"""
if kwargs:
verbose = True
warnings.warn('`full` in `show_stats` is deprecated, '
'use `verbose` instead.', DeprecationWarning)
found_proxies = self.unique_proxies.values()
num_working_proxies = len([p for p in found_proxies if p.is_working])
if not found_proxies:
print('Proxy not found')
return
errors = Counter()
for p in found_proxies:
errors.update(p.stat['errors'])
proxies_by_type = {'SOCKS5': [], 'SOCKS4': [], 'HTTPS': [], 'HTTP': [],
'CONNECT:80': [], 'CONNECT:25': []}
stat = {'Wrong country': [],
'Wrong protocol/anonymity lvl': [],
'Connection success': [],
'Connection timeout': [],
'Connection failed': []}
for p in found_proxies:
msgs = ' '.join([l[1] for l in p.get_log()])
full_log = [p, ]
for proto in p.types:
proxies_by_type[proto].append(p)
if 'Location of proxy' in msgs:
stat['Wrong country'].append(p)
elif 'Connection: success' in msgs:
if 'Protocol or the level' in msgs:
stat['Wrong protocol/anonymity lvl'].append(p)
stat['Connection success'].append(p)
if not verbose:
continue
events_by_ngtr = defaultdict(list)
for ngtr, event, runtime in p.get_log():
events_by_ngtr[ngtr].append((event, runtime))
for ngtr, events in sorted(events_by_ngtr.items(),
key=lambda item: item[0]):
full_log.append('\t%s' % ngtr)
for event, runtime in events:
if event.startswith('Initial connection'):
full_log.append('\t\t-------------------')
else:
full_log.append('\t\t{:<66} Runtime: {:.2f}'
.format(event, runtime))
for row in full_log:
print(row)
elif 'Connection: failed' in msgs:
stat['Connection failed'].append(p)
else:
stat['Connection timeout'].append(p)
if verbose:
print('Stats:')
pprint(stat)
print('The number of working proxies: %d' % num_working_proxies)
for proto, proxies in proxies_by_type.items():
print('%s (%s): %s' % (proto, len(proxies), proxies))
print('Errors:', errors)
def _update_types(types):
_types = {}
if not types:
return _types
elif isinstance(types, dict):
return types
for tp in types:
lvl = None
if isinstance(tp, (list, tuple, set)):
tp, lvl = tp[0], tp[1]
if isinstance(lvl, str):
lvl = lvl.split()
_types[tp] = lvl
return _types
