<commit_before>from typing import Callable
from fp.fp import FreeProxy
import random
import logging
import time
import requests
import stem.process
import tempfile
import os

from requests.exceptions import Timeout
from selenium import webdriver
from selenium.webdriver.support.wait import WebDriverWait, TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.common.exceptions import WebDriverException, UnexpectedAlertPresentException
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from urllib.parse import urlparse
from stem import Signal
from stem.control import Controller
from fake_useragent import UserAgent
from dotenv import load_dotenv, find_dotenv

class DOSException(Exception):
    """DOS attack was detected."""


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args,
                                                                 **kwargs)
        return cls._instances[cls]

class ProxyGenerator(object):
    def __init__(self):
        # setting up logger
        self.logger = logging.getLogger('scholarly')

        self._proxy_gen = None
        # If we use a proxy or Tor, we set this to True
        self._proxy_works = False
        self._use_luminati = False
        self._use_scraperapi = False
        # If we h:ve a Tor server that we can refresh, we set this to True
        self._tor_process = None
        self._can_refresh_tor = False
        self._tor_control_port = None
        self._tor_password = None
        self._session = None
        self._TIMEOUT = 5
        self._new_session()

    def __del__(self):
        if self._tor_process:
            self._tor_process.kill()
            self._tor_process.wait()
        self._close_session()

    def get_session(self):
        return self._session

    def Luminati(self, usr , passwd, proxy_port):
        """ Setups a luminati proxy without refreshing capabilities.

        :param usr: scholarly username, optional by default None
        :type usr: string
        :param passwd: scholarly password, optional by default None
        :type passwd: string
        :param proxy_port: port for the proxy,optional by default None
        :type proxy_port: integer
        :returns: whether or not the proxy was set up successfully
        :rtype: {bool}

        :Example::
            pg = ProxyGenerator()
            success = pg.Luminati(usr = foo, passwd = <PASSWORD>, port = 1200)
        """
        if (usr != None and passwd != None and proxy_port != None):
            username = usr
            password = <PASSWORD>
            port = proxy_port
        else:
            self.logger.info("Not enough parameters were provided for the Luminati proxy. Reverting to a local connection.")
            return
        session_id = random.random()
        proxy = f"http://{username}-session-{session_id}:{password}@zproxy.lum-superproxy.io:{port}"
        proxy_works = self._use_proxy(http=proxy, https=proxy)
        return proxy_works

    def SingleProxy(self, http = None, https = None):
        """
        Use proxy of your choice
        :param http: http proxy address
        type http: string
        :param https: https proxy adress
        :type https: string
        :returns: whether or not the proxy was set up successfully
        :rtype: {bool}

        :Example::
            pg = ProxyGenerator()
            success = pg.SingleProxy(http = <http proxy adress>, https = <https proxy adress>)
        """
        proxy_works = self._use_proxy(http=http,https=https)
        return proxy_works

    def _check_proxy(self, proxies) -> bool:
        """Checks if a proxy is working.
        :param proxies: A dictionary {'http': url1, 'https': url1}
                        with the urls of the proxies
        :returns: whether the proxy is working or not
        :rtype: {bool}
        """
        with requests.Session() as session:
            session.proxies = proxies
            try:
                resp = session.get("http://httpbin.org/ip", timeout=self._TIMEOUT)
                if resp.status_code == 200:
                    self.logger.info("Proxy works! IP address: %s",
                                     resp.json()["origin"])
                    return True
            except Exception as e:
                self.logger.info(f"Exception while testing proxy: {e}")

            return False

    def _refresh_tor_id(self, tor_control_port: int, password: str) -> bool:
        """Refreshes the id by using a new Tor node.

        :returns: Whether or not the refresh was succesful
        :rtype: {bool}
        """
        try:
            with Controller.from_port(port=tor_control_port) as controller:
                if password:
                    controller.authenticate(password=password)
                else:
                    controller.authenticate()
                controller.signal(Signal.NEWNYM)
                self._new_session()
            return (True, self._session)
        except Exception as e:
            err = f"Exception {e} while refreshing TOR. Retrying..."
            self.logger.info(err)
            return (False, None)

    def _use_proxy(self, http: str, https: str = None) -> bool:
        """Allows user to set their own proxy for the connection session.
        Sets the proxy, and checks if it works.

        :param http: the http proxy
        :type http: str
        :param https: the https proxy (default to the same as http)
        :type https: str
        :returns: whether or not the proxy was set up successfully
        :rtype: {bool}
        """
        if https is None:
            https = http

        proxies = {'http': http, 'https': https}
        self._proxy_works = self._check_proxy(proxies)
        if self._proxy_works:
            self.logger.info(f"Enabling proxies: http={http} https={https}")
            self._session.proxies = proxies
            self._new_session()
            # check if the proxy url contains luminati or scraperapi
            self._use_luminati = (True if "lum" in http else False)
            self._use_scraperapi = (True if "scraperapi" in http else False)
        else:
            self.logger.info(f"Proxy {http} does not seem to work.")
        return self._proxy_works

    def Tor_External(self, tor_sock_port: int, tor_control_port: int, tor_password: str):
        """
        Setting up Tor Proxy. A tor service should be already running on the system. Otherwise you might want to use Tor_Internal

        :param tor_sock_port: the port where the Tor sock proxy is running<commit_msg>Change logging level if proxy fails<commit_after>from typing import Callable
from fp.fp import FreeProxy
import random
import logging
import time
import requests
import stem.process
import tempfile
import os

from requests.exceptions import Timeout
from selenium import webdriver
from selenium.webdriver.support.wait import WebDriverWait, TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.common.exceptions import WebDriverException, UnexpectedAlertPresentException
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from urllib.parse import urlparse
from stem import Signal
from stem.control import Controller
from fake_useragent import UserAgent
from dotenv import load_dotenv, find_dotenv

class DOSException(Exception):
    """DOS attack was detected."""


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args,
                                                                 **kwargs)
        return cls._instances[cls]

class ProxyGenerator(object):
    def __init__(self):
        # setting up logger
        self.logger = logging.getLogger('scholarly')

        self._proxy_gen = None
        # If we use a proxy or Tor, we set this to True
        self._proxy_works = False
        self._use_luminati = False
        self._use_scraperapi = False
        # If we h:ve a Tor server that we can refresh, we set this to True
        self._tor_process = None
        self._can_refresh_tor = False
        self._tor_control_port = None
        self._tor_password = None
        self._session = None
        self._TIMEOUT = 5
        self._new_session()

    def __del__(self):
        if self._tor_process:
            self._tor_process.kill()
            self._tor_process.wait()
        self._close_session()

    def get_session(self):
        return self._session

    def Luminati(self, usr , passwd, proxy_port):
        """ Setups a luminati proxy without refreshing capabilities.

        :param usr: scholarly username, optional by default None
        :type usr: string
        :param passwd: <PASSWORD>, optional by default None
        :type passwd: string
        :param proxy_port: port for the proxy,optional by default None
        :type proxy_port: integer
        :returns: whether or not the proxy was set up successfully
        :rtype: {bool}

        :Example::
            pg = ProxyGenerator()
            success = pg.Luminati(usr = foo, passwd = <PASSWORD>, port = 1200)
        """
        if (usr != None and passwd != None and proxy_port != None):
            username = usr
            password = <PASSWORD>
            port = proxy_port
        else:
            self.logger.info("Not enough parameters were provided for the Luminati proxy. Reverting to a local connection.")
            return
        session_id = random.random()
        proxy = f"http://{username}-session-{session_id}:{password}@zproxy.lum-superproxy.io:{port}"
        proxy_works = self._use_proxy(http=proxy, https=proxy)
        return proxy_works

    def SingleProxy(self, http = None, https = None):
        """
        Use proxy of your choice
        :param http: http proxy address
        type http: string
        :param https: https proxy adress
        :type https: string
        :returns: whether or not the proxy was set up successfully
        :rtype: {bool}

        :Example::
            pg = ProxyGenerator()
            success = pg.SingleProxy(http = <http proxy adress>, https = <https proxy adress>)
        """
        proxy_works = self._use_proxy(http=http,https=https)
        return proxy_works

    def _check_proxy(self, proxies) -> bool:
        """Checks if a proxy is working.
        :param proxies: A dictionary {'http': url1, 'https': url1}
                        with the urls of the proxies
        :returns: whether the proxy is working or not
        :rtype: {bool}
        """
        with requests.Session() as session:
            session.proxies = proxies
            try:
                resp = session.get("http://httpbin.org/ip", timeout=self._TIMEOUT)
                if resp.status_code == 200:
                    self.logger.info("Proxy works! IP address: %s",
                                     resp.json()["origin"])
                    return True
            except Exception as e:
                self.logger.warning("Exception while testing proxy: %s", e)

            return False

    def _refresh_tor_id(self, tor_control_port: int, password: str) -> bool:
        """Refreshes the id by using a new Tor node.

        :returns: Whether or not the refresh was succesful
        :rtype: {bool}
        """
        try:
            with Controller.from_port(port=tor_control_port) as controller:
                if password:
                    controller.authenticate(password=password)
                else:
                    controller.authenticate()
                controller.signal(Signal.NEWNYM)
                self._new_session()
            return (True, self._session)
        except Exception as e:
            err = f"Exception {e} while refreshing TOR. Retrying..."
            self.logger.info(err)
            return (False, None)

    def _use_proxy(self, http: str, https: str = None) -> bool:
        """Allows user to set their own proxy for the connection session.
        Sets the proxy, and checks if it works.

        :param http: the http proxy
        :type http: str
        :param https: the https proxy (default to the same as http)
        :type https: str
        :returns: whether or not the proxy was set up successfully
        :rtype: {bool}
        """
        if https is None:
            https = http

        proxies = {'http': http, 'https': https}
        self._proxy_works = self._check_proxy(proxies)
        if self._proxy_works:
            self.logger.info("Enabling proxies: http=%s https=%s", http, https)
            self._session.proxies = proxies
            self._new_session()
            # check if the proxy url contains luminati or scraperapi
            self._use_luminati = (True if "lum" in http else False)
            self._use_scraperapi = (True if "scraperapi" in http else False)
        else:
            self.logger.warning("Proxy %s does not seem to work.", http)
        return self._proxy_works

    def Tor_External(self, tor_sock_port: int, tor_control_port: int, tor_password: str):
        """
        Setting up Tor Proxy. A tor service should be already running on the system. Otherwise you might want to use Tor_Internal

        :param tor_sock_port: the port where the Tor sock proxy is running
{'fraction_non_alphanumeric': 0.08062495393912594, 'fraction_numerical': 0.001473947969636672, 'mean_word_length': 2.152148664343786, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2109125', 'n_tokens_mistral': 3662, 'n_tokens_neox': 3538, 'n_words': 1223}