Source code for proxybroker.api

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.

# The maximum number of providers that are parsed concurrently

[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 # 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 <>`_ :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='', 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), ...}: {('', '80'), } """ log.debug('Load proxies from the raw data') if isinstance(data, io.TextIOWrapper): data = 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.port) not in self.unique_proxies: self.unique_proxies[(, 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'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)'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