-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathproxy-scraper.py
More file actions
88 lines (69 loc) · 2.76 KB
/
Copy pathproxy-scraper.py
File metadata and controls
88 lines (69 loc) · 2.76 KB
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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
import csv
import requests
from bs4 import BeautifulSoup
from multiprocessing.pool import ThreadPool
def makesoup(url): # pass url to beautifulsoup to parse html. Url is defined in menu for each site so code doesnt have to be repeated for each site
page=requests.get(url)
print(url + " scraped successfully")
return BeautifulSoup(page.text,"html.parser")
def proxyscrape(table): # scrape proxy data from table on site, add to list that was created earlier
proxies = set()
for row in table.findAll('tr'):
fields = row.findAll('td')
count = 0
proxy = ""
for cell in row.findAll('td'):
if count == 1:
proxy += ":" + cell.text.replace(' ', '')
proxies.add(proxy)
break
proxy += cell.text.replace(' ', '')
count += 1
# print (proxy)
return proxies
def scrapeproxies(url): # contains the parent table attribute where proxy data is present
soup=makesoup(url)
return proxyscrape(table = soup.find('table', attrs={'id': 'proxylisttable'}))
def proxy_test(proxy):
"""
Tests the scraped proxies and writes the working ones along with the time it took them to load the page to an .csv file
"""
# Using httpbin used to test the proxies
url = 'http://httpbin.org/ip'
proxies = {
'http': f'http://{proxy}',
'https': f'http://{proxy}',
}
try:
r = requests.get(url, proxies=proxies, timeout=TIMEOUT)
load_time = round(r.elapsed.total_seconds(), 3) # Time it took the page to load
if r.status_code == 200:
# Check the status code, if 200 write it to the file
global count_working_proxies
count_working_proxies += 1
csv_writer.writerow([proxy, load_time])
print(f'Working proxy --> {proxy}')
except requests.exceptions.Timeout:
print(f'[TIMED OUT] Proxy took too long: {proxy} ')
except Exception as e:
# print(e)
print(f'[Error occurred] Proxy not working: {proxy}')
pass
urls = ["http://sslproxies.org", "http://free-proxy-list.net", "http://us-proxy.org", "http://socks-proxy.net"]
proxies = set()
for url in urls:
new_proxies = scrapeproxies(url)
proxies.update(new_proxies)
print ("Proxies:" + str(proxies))
TIMEOUT = 5
file_path = 'working_proxies.csv'
count_working_proxies = 0
with open(file_path, 'w', newline='', errors='ignore') as f:
csv_writer = csv.writer(f)
csv_writer.writerow(['proxy', 'load time'])
try:
# Threading to speed up the proxy testing
ThreadPool(processes=10).map(proxy_test, proxies)
except Exception as e:
print(f'[Thread error] {e}')
print(f'Total amount of working proxies {count_working_proxies}')