如何使用 Python 对代理的上下行带宽测速进行测速

Python实现检测代理IP是否可以翻墙
转载 &更新时间:日 09:12:12 & 投稿:junjie
这篇文章主要介绍了Python实现检测代理IP是否可以翻墙,本文直接给出实现代码,需要的朋友可以参考下
那堵墙着实可恨!身处IT这个圈子,经常需要用gg查资料(你也可以用来访问1024,^_^...)。当然,你也可以用百度。其实也不是我不爱用百度,是有缘由的,且听我细细道来。有一次闲得蛋疼,想看看会不会有人抄袭我的博客(尽管博客学得不咋地),于是百度了一下,结果是惊人的。我发现我自己写的博客,即使是拿整个标题去搜索,往往搜不到,搜到的是一堆爬虫爬去的结果。具体是哪些,这里就不说了,各自可以拿自己的博客试一下。以前总是手工收集几个IP用一段时间,失效了以后再重新收集几个,如此反复,烦!于是,想着写个爬虫抓取代理IP,然后每次直接数据库里面找几个出来用就行了。然而,爬虫爬过来的IP有很多已经失效了。这时又沦落为手工测试,这不是为自己增添更多的烦恼吗?于是写个检测代理IP是否可用的程序,让程序帮我检测好了。这样每次我就可以拿到可用的代理IP了。由于爬虫是用scrapy写的,为了方便维护,IP检测就作为scrapy爬虫里面的一部分好了。于是有了下面这段检测的程序:
1、创建文件:checkproxy.py
#coding=utf-8
import urllib2
import urllib
import time
import socket
ip_check_url = 'http://www.google.com.hk/'
user_agent = 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:12.0) Gecko/ Firefox/12.0'
socket_timeout = 30
# Check proxy
def check_proxy(protocol, pip):
proxy_handler = urllib2.ProxyHandler({protocol:pip})
opener = urllib2.build_opener(proxy_handler)
# opener.addheaders = [('User-agent', user_agent)] #这句加上以后无法正常检测,不知道是什么原因。
urllib2.install_opener(opener)
req = urllib2.Request(ip_check_url)
time_start = time.time()
conn = urllib2.urlopen(req)
# conn = urllib2.urlopen(ip_check_url)
time_end = time.time()
detected_pip = conn.read()
proxy_detected = True
except urllib2.HTTPError, e:
print "ERROR: Code ", e.code
return False
except Exception, detail:
print "ERROR: ", detail
return False
return proxy_detected
def main():
socket.setdefaulttimeout(socket_timeout)
protocol = "http"
current_proxy = "212.82.126.32:80"
proxy_detected = check_proxy(protocol, current_proxy)
if proxy_detected:
print (" WORKING: " + current_proxy)
print " FAILED: %s " % ( current_proxy, )
if __name__ == '__main__':
[root@bogon proxyipspider]# python checkproxy.py
WORKING: 212.82.126.32:80
  当然,这只是程序的一个原型,真正检测的程序还需要结合数据库或文件操作来完成。代理IP检测出来了,那么剩下的就是设置了。设置好以后,尽情地gg吧。1024你想看多久就看多久,不过还是不要看多为好,你懂的。如果你想上脸谱、油土鳖和推特,这些就靠你自己了,这里只是gg而已。
  程序猿嘛,总是想通过自己的双手来解决点问题。那颗change the world的心没有变,就像博客园的口号那样“代码改变世界”。看到哪个东西不爽,自己造一个吧。IT界这样的例子太多了,天天用的Vi、github等等。好了,就到这,1024去,走起。。。
  那道墙着实可恨!
您可能感兴趣的文章:
大家感兴趣的内容
12345678910
最近更新的内容
常用在线小工具&nbsp>&nbsp
&nbsp>&nbsp
&nbsp>&nbsp
python测试代理速度源代码
摘要:#!/usr/bin/envpython&&&Getproxiesfromurls,andtesttheirspeed&&&importurllib,re,time,threadingurls=[&http://proxy.ipcn.org/proxylist.html&,&http://info.hustonline.net/index/proxyshow.aspx&]#wh
#!/usr/bin/env python
&&&Get proxies from urls, and test their speed&&&
import urllib, re, time, threading
urls = [&http://proxy.ipcn.org/proxylist.html&,
&http://info.hustonline.net/index/proxyshow.aspx&
]#where to get proxies
urls_proxy = {}#proxy used to connect urls
proxy_pattern = re.compile(r&&&d{1,3}.d{1,3}.d{1,3}.d{1,3}:d{1,}&&&)
test_url =&http://www.python.org/&
test_pattern = re.compile(r&&&xs4all&&&)
time_out = 30.0#max waiting time to test proxies
output_file =&Proxies.txt&
class TestTime(threading.Thread):
&&&test a proxy's speed in new thread by recording its connect time&&&
def __init__(self, proxy):
threading.Thread.__init__(self)
self.proxy = proxy
self.time = None
self.stat = proxy +&time out!&
def run(self):
start = time.time()
f = urllib.urlopen(test_url, proxies = {&http&:&http://&+self.proxy})
self.stat = self.proxy+&fails!&
data = f.read()
end = time.time()
if test_pattern.search(data): #if data is matched
self.time = end-start
self.stat = self.proxy+&time:&+str(self.time)
self.stat = self.proxy+&not matched!&
def totest(proxy, result):
&&&test a proxy's speed in time_out seconds&&&
test = TestTime(proxy)
test.setDaemon(True)
print&testing&+proxy
test.start()
test.join(time_out) #wait time_out seconds for testing
print test.stat
if test.time:
result.append((test.time, proxy))
if __name__ ==&__main__&:
#get old proxies in output_file
f = open(output_file)
allproxies = set()
allproxies = set([x[:-1] for x in f.readlines()])
#get else proxies from urls
for url in urls:
print&getting proxy from&+url
f = urllib.urlopen(url, proxies=urls_proxy)
print url+&can not open!n&
data = f.read()
allproxies.update(proxy_pattern.findall(data))
print url+&finished!&
#test all proxies' speed
result = []
for proxy in allproxies:
#new thread to test every proxy
t = threading.Thread(target=totest, args=(proxy, result))
t.setDaemon(True)
#show all proxies' speed
time.sleep(time_out+5.0)
result.sort()
for i in xrange(len(result)):
print str(i+1)+&t&+result[i][1]+&t:t&+str(result[i][0])
#output needed proxies
num = min(abs(int(raw_input(&nHow many proxies to output:&))), len(result))
f = open(output_file,&w&)
print&Can not open output file!&
f.writelines([x[1]+&n&for x in result[:num]])
print str(num)+&proxies are output.&
以上是的内容,更多
的内容,请您使用右上方搜索功能获取相关信息。
若你要投稿、删除文章请联系邮箱:zixun-group@service.aliyun.com,工作人员会在五个工作日内给你回复。
云服务器 ECS
可弹性伸缩、安全稳定、简单易用
&40.8元/月起
预测未发生的攻击
&24元/月起
邮箱低至5折
推荐购买再奖现金,最高25%
&200元/3月起
你可能还喜欢
你可能感兴趣
阿里云教程中心为您免费提供
python测试代理速度源代码相关信息,包括
的信息,所有python测试代理速度源代码相关内容均不代表阿里云的意见!投稿删除文章请联系邮箱:zixun-group@service.aliyun.com,工作人员会在五个工作日内答复
售前咨询热线
支持与服务
资源和社区
关注阿里云
International利用Python进行Web渗透测试(五):剖析HTTP请求
我的图书馆
利用Python进行Web渗透测试(五):剖析HTTP请求
本篇概要URL的结构HTTP头域HTTP状态码一个完整的HTTPget请求在上一篇我们简单介绍了HTTP协议的概念和知识,本篇我们来详细剖析一下HTTP的请求,将这两篇综合起来,我们就对HTTP有了基本的认识和了解了。&URL结构解析通常情况下,我们发起一个HTTP请求都是通过访问一个URL网址来实现。比如我们打开百度的首页(http://www.baidu.com),就是对百度的服务器发起了一个HTTP的请求。网址在我们的互联网生活中,很常见,那网址的结构又有什么讲究呢?我们来看一个典型的URL——百度搜索“州的先生”:https://www.baidu.com/s?wd=州的先生这其中:https——表示HTTP请求的使用的协议,这里使用的是https;www.baidu.com——表示服务器端的域名地址;s——表示我们的资源的路径,这里的s应当是搜索资源的路径;wd=州的先生——这就是我们发送的请求的查询字符串;&HTTP头域HTTP的头域描述和客户端和服务器端如何相互进行通信,并且提供了相关的事务信息。HTTP的头域包括了请求头和响应头。HTTP响应的头域Connection:表示这个HTTP连接的模式,前面我们提到,HTTP的运行模式是客户端进行请求,服务器端进行响应,默认情况下,Connection的值为close,当这个过程完成,HTTP连接便断开。如果使用了keep-alive值,Keep-Alive功能能使客户端到服务器端的连接持续有效,当出现对服务器的后继请求时,Keep-Alive功能避免了建立或者重新建立连接。Content-Encoding:表示响应实体的编码方式。浏览器在发送请求的时候,会通过 Accept-Encoding头域 带上自己支持的内容编码格式列表;服务器端接收到之后,会从中挑选一种用来对响应实体进行编码,并通过 Content-Encoding 响应头指明选定的格式;浏览器拿到响应正文后,依据 Content-Encoding 进行解压。Content-Type:表示响应实体的类型,用于定义响应的文件类型和网页的编码,以此来决定浏览器以什么形式、什么编码来读取这个响应实体。Date:表示响应的时间。Server:表示响应服务器的架构。Transfer-Encoding:表示响应实体的传输编码模式,图中的'chunked'代表的含义是分块传输。HTTP请求的头域Accept:表示浏览器声明请求接收的文件类型。Accept-Encoding:表示浏览器声明接受的响应内容的编码格式。Accept-Language:表示浏览器声明其接受的自然语言类型。Cache-Control:用于浏览器控制网页的缓存。其各类值指示的是浏览器再次请求网页的时候,是使用缓存,还是向服务器发出请求。Connection:表示浏览器声明此次请求的连接模式。Cookie:浏览器发出的为了声明本次请求的身份、便于进行 session 跟踪而储存在本地的数据,通常由键值对组成,常用于用户身份的持久化认证。Host:表示浏览器请求的主域名。User-Agent:表示浏览器的标识。不同操作系统、不同版本、不同厂商的浏览器的标识各不一致,通过修改这个参数可以达到爬虫初级伪装的效果。&HTTP的状态码HTTP的状态码是一个由三个十进制的数组成的代码,这个简短的三位数,可以很直观的告诉我们,这次HTTP的请求的结果是什么。HTTP的状态码由三个十进制的数表示,状态码的类型由第一位数来决定。HTTP的状态码一共分为五种类型:1xx:表示服务器收到请求,需要客户端继续执行操作。2xx:表示请求成功,并且服务器对请求做出了响应。3xx:表示重定向。4xx:表示客户端请求错误。5xx:表示服务器错误。在实际的应用当中,我们经常遇到的状态码有以下几种:200:请求成功304:重定向400:客户端请求语法错误403:客户端请求被拒绝,可能是权限问题404:请求的资源没有找到500:服务器内部错误502:网关或代理服务器错误504:网关超时,服务器没有及时从上游服务器接获取到全球&一个GET请求GET请求方法是显式的请求方法,通过URL参数,将需要的所有信息发送给URL。例如上一节我们通过百度搜索“州的先生”就是向百度的服务器发送了一个GET请求。下面,我们以www.httpbin.org这个网站作为测试示例。httpbin.org是一个测试HTTP请求和响应的网站,基于Python + Flask 平台,能够测试 HTTP 请求和响应的各种信息,比如 cookie、ip、headers 和登录验证等,且支持 GET、POST 等多种方法,对 web 开发和测试很有帮助。&一个成功的请求我们点击www.httpbin.org页面上的/ip链接,页面会跳转到返回一个服务器的IP地址:我们打开浏览器控制台的网络选项卡看看,其中:Headers:表示的就是头域了,其中包含了请求头和响应头。Preview:表示预览请求资源的响应实体。Response:请求资源的响应实体。Cookies:记录浏览信息或者用于认证的数据Timing:请求这个资源所花费的时间里面详细的显示了整个HTTP的请求和响应的信息。我们看看Headers选项卡里面的General部分,主要概述本次HTTP连接的总体信息。Request URL,表示请求的URL地址;Request Method,表示请求的方法,GET;Status Code,表示响应的状态码,200;Remote Address,表示响应服务器的IP地址。其余的Response Headers部分和Requests Headers部分就是我们之前介绍的响应头域和请求头域,其中的内容大都能找到代表的意思。这样,我们就完成了对一个基本的HTTP请求剖析。下一篇,我们介绍使用Python的模块进行HTTP请求。(类似于爬虫的……)如果有疑问,欢迎留言提问。
[转]&[转]&
喜欢该文的人也喜欢Python 批量获取代理服务器IP并测试
import urllib2
import BeautifulSoup
import codecs
User_Agent = 'Mozilla/5.0 (Windows NT 6.3; WOW64; rv:43.0) Gecko/ Firefox/43.0'
header = {}
header['User-Agent'] = User_Agent
url = 'http://www.xicidaili.com/nn/1'
req = urllib2.Request(url,headers=header)
res = urllib2.urlopen(req).read()
soup = BeautifulSoup.BeautifulSoup(res)
ips = soup.findAll('tr')
f = codecs.open("./proxy","w", 'utf-8')
for x in range(1, len(ips)):
ip = ips[x]
tds = ip.findAll("td")
ip_temp = tds[1].contents[0]+"\t"+tds[2].contents[0]+"\n"
f.write(ip_temp)
import urllib
import socket
socket.setdefaulttimeout(3)
f = open("./proxy")
fd_proxy = codecs.open("./access.txt", "w", 'utf-8')
lines = f.readlines()
proxys = []
for i in range(0, len(lines)):
ip = lines[i].strip("\n").split("\t")
proxy_host = "http://" + ip[0] + ":" + ip[1]
proxy_temp = {"http":proxy_host}
proxys.append(proxy_temp)
url = "http://ip.chinaz.com/getip.aspx"
for proxy in proxys:
res = urllib.urlopen(url,proxies=proxy).read()
fd_proxy.write(proxy["http"]+"\n")
except Exception,e:
print proxy
fd_proxy.close()1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484912345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
我们获取的代理服务器网址位于
里面会经常更新可用的代理服务器ip,如图所示
在safari浏览器页面右键弹出快捷菜单,选择显示页面源文件就能看到如下界面:
我们先获得tr字段,然后从tr字段中选择ip地址所在的td字段就能把代理服务器地址给提取出来,同样的方法提取端口号,然后保存为文件proxy。
保存下来后,不能保证每一个代理服务器地址都是可以连接的,需要进行,后半部分代码就是把保存在proxy文件中的ip地址一个个拿出来,通过访问页面
进行测试。
没有更多推荐了,
加入CSDN,享受更精准的内容推荐,与500万程序员共同成长!思路:利用ruquest发送请求,利用多线程模拟并发
下面直接上代码:
#!/user/bin/env python
#coding=utf-<span style="color: #
import requests
import datetime
import time
import threading
class url_request():
times = []
error = []
def req(self,AppID,url):
myreq=url_request()
headers = {'User-Agent' : 'Mozilla/5.0 (L Android 4.2.1; en- Nexus 4 Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0. Mobile Safari/535.19'}
payload = {'AppID':AppID,'CurrentURL':url}
r = requests.post("http://xx.xxx.com/WeiXinJSAccessToken/json/WeChatJSTicket",headers=headers,data=payload)
ResponseTime=float(r.elapsed.microseconds)/<span style="color: #00 #获取响应时间,单位ms
myreq.times.append(ResponseTime) #将响应时间写入数组
if r.status_code !=<span style="color: #0 :
myreq.error.append("<span style="color: #")
if __name__=='__main__':
myreq=url_request()
threads = []
starttime = datetime.datetime.now()
print "request start time %s" %starttime
nub = <span style="color: ##设置并发线程数
ThinkTime = <span style="color: #.5#设置思考时间
for i in range(<span style="color: #, nub+<span style="color: #):
t = threading.Thread(target=myreq.req, args=('<span style="color: #','http://m.ctrip.com/webapp/cpage/#mypoints'))
threads.append(t)
for t in threads:
time.sleep(ThinkTime)
#print "thread %s" %t #打印线程
t.setDaemon(True)
endtime = datetime.datetime.now()
print "request end time %s" %endtime
time.sleep(<span style="color: #)
AverageTime = "{:.3f}".format(float(sum(myreq.times))/float(len(myreq.times))) #计算数组的平均值,保留3位小数
print "Average Response Time %s ms" %AverageTime #打印平均响应时间
usetime = str(endtime - starttime)
hour = usetime.split(':').pop(<span style="color: #)
minute = usetime.split(':').pop(<span style="color: #)
second = usetime.split(':').pop(<span style="color: #)
totaltime = float(hour)*<span style="color: #*<span style="color: # + float(minute)*<span style="color: # + float(second) #计算总的思考时间+请求时间
print "Concurrent processing %s" %nub #打印并发数
print "use total time %s s" %(totaltime-float(nub*ThinkTime)) #打印总共消耗的时间
print "fail request %s" %myreq.error.count("<span style="color: #") #打印错误请求数
request start time <span style="color: #15-<span style="color: #-<span style="color: # <span style="color: #:<span style="color: #:<span style="color: #.316000
request end time <span style="color: #15-<span style="color: #-<span style="color: # <span style="color: #:<span style="color: #:<span style="color: #.769000
Average Response Time <span style="color: #.700 ms
Concurrent processing <span style="color: #
use total time <span style="color: #.453 s
fail request <span style="color: #
还可以据此计算tps,也可以控制并发量循环找出符合响应时间要求的最大并发量,等等
阅读(...) 评论()}

我要回帖

更多关于 带宽测速 的文章

更多推荐

版权声明:文章内容来源于网络,版权归原作者所有,如有侵权请点击这里与我们联系,我们将及时删除。

点击添加站长微信