idandaniel / ballyregan Goto Github PK

from dataclasses import dataclass
from typing import List

from ballyregan import Proxy
from ballyregan.models import Protocols
from ballyregan.providers import IProxyProvider
from ballyregan.core.exceptions import ProxyGatherException

@dataclass
class ProxyscrapeProvider(IProxyProvider):
    # https://docs.proxyscrape.com/
    url: str = 'https://api.proxyscrape.com/v2/?request=getproxies&country=all&ssl=all'

    def _get_raw_proxies(self) -> List[str]:
        proxies = []

        for anonymity in ['elite', 'transparent', 'anonymous']:
            for protocol in Protocols.values():
                try:
                    proxies_response = self._session.get(f'{self.url}', params={'protocol': protocol, 'anonymity': anonymity})
                    if not proxies_response.ok:
                        raise ProxyGatherException
                except (IndexError, ValueError, ConnectionError) as e:
                    raise ProxyGatherException from e
                else:
                    for proxy in proxies_response.text.splitlines():
                        ip, port = proxy.split(':')
                        proxies.append({
                            'protocol': protocol,
                            'ip': ip,
                            'port': port,
                            'anonymity': anonymity
                        })

        return proxies

    @staticmethod
    def raw_proxy_to_object(raw_proxy: dict) -> Proxy:
        return Proxy(
            protocol=raw_proxy['protocol'],
            ip=raw_proxy['ip'],
            port=raw_proxy['port'],
            anonymity=raw_proxy['anonymity']
        )

def test_proxyscrapeprovider():
    fetcher = ProxyFetcher(debug=False)
    fetcher._proxy_providers = [ProxyscrapeProvider()]

    proxies = fetcher._get_all_proxies_from_providers()
    assert len(proxies) > 0

    valid_proxies = fetcher._proxy_validator.filter_valid_proxies(proxies)
    assert len(valid_proxies) > 0

import random
from aiohttp import web, ClientSession
from ballyregan import ProxyFetcher
from ballyregan.models import Protocols

# A list to hold our proxies and number of uses
proxies = []

MAX_USES = 10  # Maximum number of uses per proxy
REFETCH_NO = 20  # Number of proxies you want to fetch initially and when the list is exhausted

# Prepare the proxy fetcher
fetcher = ProxyFetcher(debug=True)

# Fetch and format proxies
def fetch_proxies():
    proxies.clear()
    raw_proxies = fetcher.get(
        limit=REFETCH_NO,
        protocols=[Protocols.HTTPS, Protocols.SOCKS4, Protocols.SOCKS5]
    )
    for p in raw_proxies:
        proxies.append({
            "proxy": f'{p.ip}:{p.port}',
            "type": {'https': 'HTTPS', 'socks4': 'SOCKS4', 'socks5': 'SOCKS5'}.get(p.protocol.lower()),
            "count": 0,
        })
    print(f'Fetched {len(proxies)} proxies. Proxy list: {proxies}')


async def handle_request(request):
    # Prepare headers for forward request
    headers = {k: v for k, v in request.headers.items()}
    url = 'https://hidden.due/to/privacy/reasons'
    while True:
        if not proxies:  # If no proxies left
            fetch_proxies()

        cur_proxy = random.choice(proxies)
        count = cur_proxy["count"]

        if count >= MAX_USES: # If proxy use count exceed the maximum use limit
            proxies.remove(cur_proxy) # If so, remove the proxy from the list
        else:
            try:
                cur_proxy["count"] += 1
                async with ClientSession() as session:
                    # Correctly format the proxy URL depending on the type
                    proxy_url = {'HTTP': 'http://', 'HTTPS': 'https://', 'SOCKS4': 'socks4://', 'SOCKS5': 'socks5://'}.get(cur_proxy['type'], 'http://') + cur_proxy["proxy"]
                    print(f'Proxy url is {proxy_url}')
                    async with session.request(
                        request.method,
                        url,
                        headers=headers,
                        allow_redirects=False,
                        data=await request.read(),
                        proxy=proxy_url
                    ) as resp:
                        resp_headers = {k: v for k, v in resp.headers.items()}
                        body = await resp.text()
                        break
            except:
                proxies.remove(cur_proxy)  # If the proxy is not working, remove it from the list

    return web.Response(
        headers=resp_headers,
        status=resp.status,
        body=body
    )

app = web.Application()
app.router.add_route('*', '/{tail:.*}', handle_request)
fetch_proxies()  # Fetch proxies initially

try:
    web.run_app(app, port=1337)
except KeyboardInterrupt:
    print("KeyboardInterrupt received, exiting...")

@dataclass
class TestProvider(IProxyProvider):
   def _get_raw_proxies(self,) -> List[Any]:
      # Do some stuff

   def raw_proxy_to_object(raw_proxy: Any) -> Proxy:
      # Do some stuff

proxies = fetcher.get(
  timeout=3000
)
print(proxies)

    fetcher = ProxyFetcher(debug=False)
    proxies = fetcher.get(protocols=[Protocols.HTTPS])

    with requests.session() as reqs:
        for idx in range(len(proxies)):
            reqs.proxies.update({
                "http": str(proxies[idx]),
                "https": str(proxies[idx]),
            })
            try:
                response = reqs.get('https://api.ipify.org', timeout=15, verify=False)
                #response = reqs.get('http://httpheader.net/azenv.php', timeout=15, verify=False)
                assert response.status_code == 200
                print(f"INFO Proxy: {proxies[idx]} is valid")
            except Exception as e:
                print(f"WARNING Proxy: {proxies[idx]} is invalid. Error:{str(e)}")