panxy/panxy.py

#!/usr/bin/env python3
# panxy, a random proxy connector

from urllib import request, error 
import socket
import sys
target = "http://theresno.cloud/ip"
prx_api = "https://www.proxyscan.io/api/proxy?last_check=9800&limit=10&type=http" #"http://pubproxy.com/api/proxy?limit=10&http=true&format=txt"

def get_list_from_api():
    import json
    print('Requesting proxy list... ', end='')
    result = request.urlopen(prx_api)
    # json magic starts here
    content = result.read().decode()
    json_data = json.loads(content)
    prx_list = ["".join(json_data[i]["Ip"] + ':' + str(json_data[i]["Port"])) for i in range(len(json_data))]
    #= result.read().decode().split('\n')
    print('got {} proxies.'.format(len(prx_list)))
    return prx_list

def test_proxy(prx):
    try:
        from http.client import RemoteDisconnected
    except ImportError:
        from httplib import BadStatusLine as RemoteDisconnected
    srv_string = "Testing proxy " + prx + "... "
    print(srv_string, end='\r')
    proxy_handler = request.ProxyHandler({'https': 'http://' + prx + '/', 'http': 'http://' + prx + '/' })
    opener = request.build_opener(proxy_handler)
    try:
        result = opener.open(target, timeout=6)
        content = result.read().decode()
        if prx.split(':')[0] not in content:
            print(srv_string + 'bait.')
            return False, prx
        print(srv_string + 'ok. \u2705')
        return True, prx
    except error.HTTPError:
        print(srv_string + 'fail. \u274c')
        return False, prx
    except error.URLError:
        print(srv_string + 'down! \u2935')
        return False, prx
    except RemoteDisconnected:
         print(srv_string + 'connection fail!')
         return False, prx
    except socket.timeout:
        print(srv_string + 'timeout! \u23f3')
        return False, prx
       
def get_proxies():
    #from threading import Thread
    import concurrent.futures
    proxy_list = get_list_from_api()
    working_list = []
    start_length = len(proxy_list)
    threads = []
    print('Got {} proxies.'.format(len(proxy_list)))
    try:
        with concurrent.futures.ThreadPoolExecutor() as executor:
            stuff = [executor.submit(test_proxy, entry) for entry in proxy_list if entry not in working_list]
        for f in stuff:
            result = f.result()
            if result[0]:
                working_list.append(result[1])
    except KeyboardInterrupt:
        print(' Impatient? Interrupt again to terminate')

    finally:
        print('{}/{} new proxies ready.'.format(len(working_list), start_length))
        return working_list

def user_loop():
    proxies = []
    while True:
        proxies += get_proxies()
        print('{} proxies total ready.'.format(len(proxies)))
        choice = input('retry or continue? r|c ')
        if choice == 'c':
            break
    print(proxies)
    return proxies

def main():
    user_loop()

if __name__ == '__main__':
  main()