设为首页 收藏本站
查看: 675|回复: 0

[经验分享] python版本的curl工具pycurl学习

[复制链接]

尚未签到

发表于 2018-8-4 08:19:11 | 显示全部楼层 |阅读模式
  
import os,sys,time
  
import threading
  
import Queue
  

  
try:
  
   from cStringIO import StringIO
  
except ImportError:
  
   from StringIO import StringIO
  
import pycurl
  

  
# We should ignore SIGPIPE when using pycurl.NOSIGNAL - see
  
# the libcurl tutorial for more info.
  
try:
  
   import signal
  
   from signal import SIGPIPE,SIG_ING
  
   signal.signal(signal.SIGPIPE,signal.SIG_IGN)
  
except ImportError:
  
   pass
  

  

  
# need a given txt file contains urls
  
try:
  
   if sys.argv[1]=="-":
  
      urls=sys.stdin.readlines()
  
   else:
  
      urls=open(sys.argv[1],'rb').readlines()
  
   #print urls
  
except:
  
   print "Usage: %s check_urls.txt <file with urls to check>" %sys.argv[0]
  
   raise SystemExit
  

  
class Curl:
  
   def __init__(self,url):
  
       self.url=url
  
       self.body=StringIO()
  
       self.http_code=0
  

  
       self._curl=pycurl.Curl()
  
       self._curl.setopt(pycurl.URL,self.url)
  
       self._curl.setopt(pycurl.FOLLOWLOCATION,True)
  
       self._curl.setopt(pycurl.CONNECTTIMEOUT,15)
  
       self._curl.setopt(pycurl.TIMEOUT,15)
  
       self._curl.setopt(pycurl.WRITEFUNCTION,self.body.write)
  
       self._curl.setopt(pycurl.NOSIGNAL,1)
  
       self._curl.debug=0
  

  
   def perform(self):
  
       self._curl.perform()
  

  
   def close(self):
  
      try:
  
        self.http_code=self._curl.getinfo(pycurl.HTTP_CODE)
  
      except pycurl.error:
  
        assert c.debug
  
        self.http_code=0
  
      self._curl.close()
  

  

  
queue=Queue.Queue()
  
for url in urls:
  
    url=url.strip()
  
    if not url or url[0] == "#":
  
       continue
  
    queue.put(url)
  

  
assert queue.queue, "no urls are given"
  
num_urls=len(queue.queue)
  
#num_conn=min(num_conn,num_urls)
  
num_conn=num_urls
  
#assert 1 <= num_conn < = 1000,"invalid number of concurrent connections"
  

  
class WorkerThread(threading.Thread):
  
     def __init__(self,queue):
  
         threading.Thread.__init__(self)
  
         self.queue=queue
  

  
     def run(self):
  
         while 1:
  
             try:
  
                url=self.queue.get_nowait()
  
             except Queue.Empty:
  
                raise SystemExit
  
             c=Curl(url)
  
             c.perform()
  
             c.close()
  
             print "http_url:" + url + "\t" + "http_code:" + str(c.http_code)
  
#start a bunch of threads
  
threads=[]
  
for dummy in range(num_conn):
  
    t=WorkerThread(queue)
  
    t.start()
  
    threads.append(t)
  

  
#wait for all threads to finish
  
for thread in threads:
  
    thread.join()

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.iyunv.com/thread-546197-1-1.html 上篇帖子: python学习之认识fabric 下篇帖子: python中文转换url编码
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表