LCB/proxy.py

import requests
import signal
from database import get_proxys
from Console import console
from settings import USE_PROXY


class Proxy:
    def __init__(self, ip: str, protocol: str, port: int):
        self.ip = ip
        self.protocol = protocol
        self.port = port

    def telethon_proxy(self):
        if self.protocol == "SOCKS5":
            return "socks5", self.ip, self.port
        elif self.protocol == "SOCKS4":
            return "socks5", self.ip, self.port
        elif self.protocol == "HTTP":
            return "http", self.ip, self.port
        elif self.protocol == "HTTPS":
            return "https", self.ip, self.port
        else:
            return None, None, None

    def selenium_proxy(self):
        if self.ip and self.port:
            return self.ip + ':' + str(self.port)
        else:
            return ""

    def proxy(self):
        return self.ip, self.protocol, self.port


class Proxy_controller:
    def __init__(self, proxy: list, for_one=5, use_proxy=USE_PROXY):
        self.list_of_proxy = []
        for el in proxy:
            self.list_of_proxy.append(Proxy(el[0], el[1], int(el[2])))

        self.pos = 0
        self.tries = 0
        self.tries_for_one = for_one
        self.use_proxy = use_proxy

    def get_proxy(self):
        if len(self.list_of_proxy) == 0 or not self.use_proxy:
            return None

        if self.tries < self.tries_for_one:
            return self.list_of_proxy[self.pos]
        else:
            if self.pos < len(self.list_of_proxy) - 1:
                self.pos += 1
            else:
                self.pos = 0
            self.tries = 0

            return self.list_of_proxy[self.pos]

    def add_proxy(self, proxy):
        self.list_of_proxy.append(proxy)

    def remove_proxy(self, proxy):
        self.list_of_proxy.remove(proxy)


def signal_handler(signum, frame):
    raise Exception("Timed out!")


def check_proxy(proxy_ip, mode, timeout=10, url="https://google.com"):
    proxy = {"http": "socks5://{}".format(proxy_ip), "https": "socks5://{}".format(proxy_ip)}
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(timeout)  # Timeout in seconds
    try:
        requests.get(url, proxies=proxy)
        return True
    except Exception as e:
        print("Exception:", e)
        return False


def available_proxy():
    proxy = get_proxys()
    res = []

    if proxy:
        for el in proxy:
            # el: ip, protocol, port
            if check_proxy(el[0], el[1]):
                res.append(el)

    return res


def get_all_proxys():
    all_proxy = get_proxys()

    ret = []
    for el in all_proxy:
        if int(el[3]):
            ret.append(Proxy(el[0], el[1], int(el[2])))
    return ret


def socks_proxy():
    ret = []
    for el in get_proxys():
        if el[1] == "SOCKS5" and el[3]:
            ret.append(Proxy(el[0], el[1], int(el[2])))
    return ret


def http_proxy():
    ret = []
    for el in get_proxys():
        if el[1] == "HTTP" and el[3]:
            ret.append(Proxy(el[0], el[1], int(el[2])))
    return ret


if __name__ == '__main__':
    c = Proxy_controller(get_proxys())
    print(c.get_proxy().telethon_proxy())