Все началось с того что мне понадобилось зарегистрироваться в соц сети www.linkedin.com, оказалось что он заблокирован у нас в стране, быстро нашлось решение — нашел сервис предоставляющие списки прокси-серверов, это помогло обойти блокировку, но оказалось что прокси живут не долго, это побудило реализовать скрипт который отправляет запросы на api сервиса и проверяет доступность интересующего сайта через конкретный прокси.
Чтобы не добавлять прокси вручную использовался paс файл, это также позволило отправить через прокси только конкретный сайт.
И так реализация:

Класс proxylist

Основная часть скрипта содержится в классе proxylist, он выполняет отправку запросов на сервис прокси листов

Инициализация

1
2
3
4
5
6
7
8
9
10
11
12

class proxylist():
def __init__(self,listurl="http://api.foxtools.ru/v2/Proxy"):
# Заполняю массив ошибок
self.error = [ True for _ in range(6) ]
self.error[0] = ''
self.listurl = listurl
# Массив аргументов
self.args = {}
self.type = {'None':0,'HTTP':1,'HTTPS':2,'SOCKS4':4, 'SOCKS5': 8,'All':15 }
self.exc = {}
# Степень важности сообщений
self.verbose = 0

При объявлении класса происходит инициализация основных переменных и массивов
переменная verbose содержит уровень важности сообщений, от 0 до 5

Коннект к сервису и отправка запроса

Отправляю запросы с помощью библиотеки requests и использую метод GET

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

def __get_request__(self):
self.args['available'] = 1
self.error[2] = "Get request %s" % (self.listurl)
if self.verbose >= 2:
print(self.error[2])
try:
# Выполняем запрос к сервису
self.req = requests.post(self.listurl, data=self.args, timeout=3)
# Блок обработки ошибок
except requests.exceptions.SSLError as e:
self.error[3] = "SSL Error for proxy list service"
except requests.exceptions.ConnectTimeout as e:
self.error[3] = "Connection timeout for proxy list service"
except requests.exceptions.ConnectionError as e:
self.error[3] = "Connection error for proxy list service"
else:
return self.req
if self.verbose >= 3:
print(self.error[3])
sys.exit(1)

Полученный ответ попадает в методы фильтрации вывода:

Фильтрация ответа

Из полученного ответа убираем прокси которые относятся к определенным странам, в которых требуемый для входа сайт точно заблокирован, также убираем типы прокси, которые нас не интересуют

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

def response(self):
resp = self.__get_request__()
return resp.json()

def set_paramer(self, param):
#Собираем все параметры в один массив
for i in param:
self.args[i] = param[i]

def set_type(self, types):
summarizedType = 0
for type in types:
summarizedType+=self.type[type]
self.set_paramer({ 'type' :summarizedType})

def __excludeFromResponse(self):
#Отсылаем запрос и выполняем фильтрацию
jsonData = self.response()['response']['items']
cleaned = [ i for i in jsonData if not i['country']['iso3166a2'] in self.exc ]
return cleaned

Также есть возможность добавлять произвольные параметры с помощью метода set_paramer
Итоговый список запишется в переменную cleaned, останется его только проверить на доступность

Проверка полученых прокси

Проверка прокси на предмет задержек, постепенно повышаем разрешенную задержку, пока не появится сервера которые станут удовлетворять условию, если вернулось несколько серверов, то берем самого молодого у которого upptime меньше всего

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

def verify(self, timeout=0.1):
verifyList = {}
proxyList = self.__excludeFromResponse()
while True:
for proxyData in proxyList:
if self.verbose >= 5:
print(self.error[5])
ip = proxyData['ip']
port = proxyData['port']
if proxyData['free'] == 'Yes' and proxyData['anonymity'] == 'HighKeepAlive':
proxies = {'https':'%s:%s'%(ip,port)}
try:
response = requests.get(url=self.url, proxies=proxies, timeout=timeout)
except requests.exceptions.ProxyError as e:
self.error[5] = "%s - %s" % (proxies['https'],e)
except requests.exceptions.SSLError as e:
self.error[5] = "%s - %s" % (proxies['https'],e)
except requests.exceptions.ConnectTimeout as e:
self.error[5] = "%s - %s" % (proxies['https'],e)
except requests.exceptions.ConnectionError as e:
self.error[5] = "%s - %s" % (proxies['https'],e)
else:
if response.status_code == 200:
verifyList[proxyData['uptime']] = {'ip': str(ip),'port': str(port)}
if len(verifyList) == 0:
# Повышение таймаута, если с предыдущим нет результатов
timeout+=0.1
self.error[2] = "Next timeout is %s" % (str(timeout))
if self.verbose >= 2:
print(self.error[2])
else:
# Сортируем по uptime и берем самый меньший
winner = sorted(verifyList)[0]
self.error[1] = "Proxy for use %s" % (verifyList[winner])
if self.verbose >= 1:
print(self.error[1])
break
return verifyList[winner]

Проверка доступности сайта

Для периодической проверки нет необходимости проверять все прокси, достаточно проверять только который был найден в предыдущий раз

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

def testProxy(self, proxy):
proxies = {'https': '%s:%s' % (proxy['ip'], proxy['port'])}
try:
response = requests.get(url=self.url, proxies=proxies, timeout=3)
except requests.exceptions.ProxyError as e:
self.error[4] = "%s - %s" % (proxies['https'],e)
except requests.exceptions.SSLError as e:
self.error[4] = "%s - %s" % (proxies['https'],e)
except requests.exceptions.ConnectTimeout as e:
self.error[4] = "%s - %s" % (proxies['https'],e)
else:
if response.status_code == 200:
self.error[1] = "Verifi is done, used proxy from cache file"
if self.verbose >= 1:
print(self.error[1])
return True
self.error[1] = "Verify is failed, start searchinng new proxy"
if self.verbose >= 1:
print(self.error[1])
if self.verbose >= 4:
print(self.error[4])
return False

В результате выполнения метод вернет True или False

Сохранение результатов

В качастве хранилища последнего прокси был выбран файл в формате json, он же является и конфигурационным файлом
Для реализации сохранения был написан класс cacheJson

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import json
import os
class cacheJson ():
def __init__(self,cacheFile):
# При инициализации считываем файл конфигурации в переменную
self.cacheFile = cacheFile
with open(self.cacheFile) as cache:
self.json = json.load(cache)
# Запись данных в файл
def saveToCacheFile(self):
with open(self.cacheFile, 'w') as writeCache:
json.dump(self.json, writeCache)
# Возращение данных по определенному домену
def data(self):
return self.json[self.domain]
# Запись данных по определенному домену
def writeItem(self,item):
self.json[self.domain] = item

В итоге класс выполняет четыре действия, считывает, записывает файл, извлекает и записывает параметры

Основная программа

Первым делом необходимо позааботиться о параметрах запуска, провожу разбор аргументов с помощью библиотеки argparse
Задаю возможный ключи: —url, —config, —pac, —template, —verbose
Параметр —template мы не рассматривали, он содержит путь к шаблону для pac файла, в качестве шаблона я использовал jinja2

Параметры командной строки

1
2
3
4
5
6
7
8
9
10
11
12
13
14

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from proxylist import proxylist
from cacheJson import cacheJson
from jinja2 import Template
import argparse

parser = argparse.ArgumentParser(description='Search proxy server and update pac file for firefox.')
parser.add_argument('-u', '--url', default="http://api.foxtools.ru/v2/Proxy", help="Url to proxy list api service")
parser.add_argument('-c', '--config', default=".cache.json", help="Path to config file in format json")
parser.add_argument('-p', '--pac', default=".proxy.pac", help="Path to output pac file for firefox")
parser.add_argument('-t', '--template', default="proxy.pac.j2", help="Path to templete for pac file")
parser.add_argument('-v', '--verbose', default=0, type=int, help="Verbose output")
args = parser.parse_args()

Поиск прокси

Выполняем проверку доступности сайта, в случае неудачи запускаем поиск.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

isDone = False
rewrite = False
proxy = proxylist(args.url)
proxy.verbose = args.verbose
if proxy.verbose > 5:
proxy.verbose = 5
cache = cacheJson(args.config)
# Для каждого домена проводи отдельные проверки
for domain in cache.json:
cache.domain = domain
proxy.url = domain
# Если сайт не доступен через сохраненный прокси, то начинаем поиск другого проки
if not proxy.testProxy(cache.data()):
proxy.set_type(['HTTPS'])
proxy.exc = ['RU']
answer = proxy.verify()
cache.writeItem(answer)
isDone = False

Если изменился хоть один прокси для какого нибудь домена, то установится флаг isDone и будет произведена перезапись PAC файла

Обновление PAC файла и файла конфигурации

Если был установлен флаг isDone, то записываем параметры в формате json и одновляем pac файл

1
2
3
4
5
6
7

if isDone or rewrite:
cache.saveToCacheFile()
html = open(args.template).read()
template = Template(html)
pacFile = template.render(block=cache.json)
with open(args.pac,'w') as pac:
pac.write(pacFile)

Шаблон PAC файла

В качестве библиотеки для создания шалобна я использовал Jinja2, так как она еще используется в ansible — это скорее проба пера, чем необходимость

1
2
3
4
5

function FindProxyForURL(url, host) {
{% for i in block %}if (shExpMatch(host, "*.{{i.split('.')[-2]}}.{{i.split('.')[-1]}}"))
return "PROXY {{block[i].ip}}:{{block[i].port}}";
{% endfor %}return "DIRECT";
}

Конфигурационный файл

Перед запуском скрипта необходимо задать конфигурацию, в качестве первого прокси можно указать 127.0.0.1:8080
Это спровоцирует поиск прокси, после чего файл перезапишется

1	{"https://www.linkedin.com": {"ip": "127.0.0.1", "port": "8080"}}

Запуск скрипта

Перед запуском скрипта необходимо установить зависимые библиотеки:

1	pip install -r requirements.txt --upgrade

Скрипт можно поместить в cron — тогда он будет запускаться по расписанию, например

1	30 /3 * * /usr/bin/python /home/user/git/proxylist/create_proxy_firefox.py --template /home/user/git/proxylist/proxy.pac.j2 --config /home/user/git/proxylist/.cache.json --pac /home/user/.proxy.pac

При запуске из терминала команда выглядит практически также:

1	python /home/user/git/proxylist/create_proxy_firefox.py --template /home/user/git/proxylist/proxy.pac.j2 --config /home/user/git/proxylist/.cache.json --verbose 5 --pac /home/user/.proxy.pac

У ключей есть значения по умолчанию, поэтому их можно опускать, если значение устраивает.

Вывод результатов работы

При максимальной выводе можно увидеть процесс поиска и будет выведен прокси «победитель»
В случае если проверка доступности прошла успешно, то выведется сообщение Check is done, used proxy from cache file
В случае недоступности сообщени вида:

1
2
3
4

Check is failed, start searchinng new proxy
127.0.0.1:8080 - HTTPSConnectionPool(host='www.linkedin.com', port=443): Max retries exceeded with url: / (Caused by ConnectTimeoutError(<urllib3.connection.VerifiedHTTPSConnection object at 0x7fba73cc07b8>, 'Connection to 127.0.0.1 timed out. (connect timeout=3)'))
Get request http://api.foxtools.ru/v2/Proxy
True

Исходный код доступн в репозитории https://github.com/CrazyBunker/proxylist

Генератор pac файлов Firefox