转python版本的curl工具pycurl学习

c.setopt(c.VERBOSE, True)

import pycurl
import re
try:
   from io import BytesIO
except ImportError:
 
   from StringIO import StringIO as BytesIO
 
headers={}
def header_function(header_line):
    # HTTP standard specifies that headers are encoded in iso-8859-1.
    # On Python 2, decoding step can be skipped.
    # On Python 3, decoding step is required.
    header_line=header_line.decode(‘iso-8859-1‘)
 
    # Header lines include the first status line (HTTP/1.x ...).
    # We are going to ignore all lines that don‘t have a colon in them.
    # This will botch headers that are split on multiple lines...
    if ‘:‘ not in header_line:
        return
 
    # Break the header line into header name and value.
    name, value = header_line.split(‘:‘, 1)
 
    # Remove whitespace that may be present.
    # Header lines include the trailing newline, and there may be whitespace
    # around the colon.
    name = name.strip()
    value = value.strip()
 
    # Header names are case insensitive.
    # Lowercase name here.
    name = name.lower()
 
    # Now we can actually record the header name and value.
    headers[name] = value
buffer=BytesIO()
c=pycurl.Curl()
c.setopt(c.URL,‘http://pycurl.io‘)
c.setopt(c.WRITEFUNCTION,buffer.write)
#set our header function
c.setopt(c.HEADERFUNCTION,header_function)
c.perform()
c.close()
 
 
# Figure out what encoding was sent with the response, if any.
# Check against lowercased header name.
encoding=None
if ‘content-type‘ in headers:
    content_type=headers[‘content-type‘].lower()
    match=re.search(‘charset=(\S+)‘, content_type)
    if match:
        encoding=match.group(1)
        print(‘Decoding using %s‘ % encoding)
 
if encoding is None:
    # Default encoding for HTML is iso-8859-1.
    # Other content types may have different default encoding,
    # or in case of binary data, may have no encoding at all.
    encoding=‘iso-8859-1‘
    print(‘Assuming encoding is %s‘ % encoding)
 
body=buffer.getvalue()
# Decode using the encoding we figured out.
print(body.decode(encoding))

import pycurl
 
with open(‘out.html‘,‘wb‘) as f:
   c=pycurl.Curl()
   c.setopt(c.URL,‘http://pycurl.io/‘)
   c.setopt(c.WRITEDATA,f)
   c.perform()
   c.close()

import pycurl
c=pycurl.Curl()
#Redirects to https://www.python.org/.
c.setopt(c.URL,‘http://www.python.org/‘)
#Follow redirect
c.setopt(c.FOLLOWLOCATION,True)
c.perform()
c.close()

import pycurl
try:
    from io import BytesIO
except ImportError:
    from StringIO import StringIO as BytesIO
 
buffer=BytesIO()
c=pycurl.Curl()
c.setopt(c.URL,‘http://www.python.org/‘)
c.setopt(c.WRITEFUNCTION,buffer.write)
c.perform()
 
#Last used URL
print(‘Effective_url: %s‘ %c.getinfo(c.EFFECTIVE_URL))
#HTTP response code
print(‘Response_code: %d‘ %c.getinfo(c.RESPONSE_CODE))
#Total time of previous transfer
print(‘Total_time: %f‘ %c.getinfo(c.TOTAL_TIME))
#Time from start until name resolving completed
print(‘Namelookup_time: %f‘ %c.getinfo(c.NAMELOOKUP_TIME))
#Time from start until remote host or proxy completed
print(‘Connect_time: %f‘ %c.getinfo(c.CONNECT_TIME))
#Time from start until SLL/SSH handshake completed
print(‘SSL/SSH_time: %f‘ %c.getinfo(c.APPCONNECT_TIME))
#Time from start until just before the transfer begins
print(‘Pretransfer_time: %f‘ %c.getinfo(c.PRETRANSFER_TIME))
#Time from start until just when the first byte is received
print(‘Starttransfer_time: %f‘ %c.getinfo(c.STARTTRANSFER_TIME))
#Time taken for all redirect steps before the final transfer
print(‘Redirect_time: %f‘ %c.getinfo(c.REDIRECT_TIME))
#Total number of redirects that were followed
print(‘Redirect_count: %d‘ %c.getinfo(c.REDIRECT_COUNT))
#URL a redirect would take you to,had you enabled redirects
print(‘Redirect_url: %s‘ %c.getinfo(c.REDIRECT_URL))
#Number of bytes uploaded
print(‘Size_upload: %d‘ %c.getinfo(c.SIZE_UPLOAD))
#Average upload speed
print(‘Speed_upload: %f‘ %c.getinfo(c.SPEED_UPLOAD))
#Number of bytes downloaded
print(‘Size_download: %d‘ %c.getinfo(c.SIZE_DOWNLOAD))
#Average download speed
print(‘Speed_download: %f‘ %c.getinfo(c.SPEED_DOWNLOAD))
 
 
 
#getinfo must be called before close
c.close()

# python response_info.py 
Effective_url: http://www.python.org/
Response_code: 301
Total_time: 0.105395
Namelookup_time: 0.051208
Connect_time: 0.078317
SSL/SSH_time: 0.000000
Pretransfer_time: 0.078322
Starttransfer_time: 0.105297
Redirect_time: 0.000000
Redirect_count: 0
Redirect_url: https://www.python.org/
Size_upload: 0
Speed_upload: 0.000000
Size_download: 0
Speed_download: 0.000000

import pycurl
try:
  #python 3
  from urllib.parse import urlencode
except ImportError:
  from urllib import urlencode
 
 
c=pycurl.Curl()
c.setopt(c.URL,‘http://pycurl.io/tests/testpostvars.php‘)
 
post_data={‘field‘:‘value‘}
#Form data must be provided already urlencoded
postfields=urlencode(post_data)
# Sets request method to POST,
# Content-Type header to application/x-www-form-urlencoded
# and data to send in request body.
c.setopt(c.POSTFIELDS, postfields)
 
c.perform()
c.close()

import pycurl
 
c = pycurl.Curl()
c.setopt(c.URL, ‘http://pycurl.io/tests/testfileupload.php‘)
 
c.setopt(c.HTTPPOST, [
    (‘fileupload‘, (
        # upload the contents of this file
        c.FORM_FILE, __file__,
    )),
])
 
c.perform()
c.close()

import pycurl
 
c = pycurl.Curl()
c.setopt(c.URL, ‘http://pycurl.io/tests/testfileupload.php‘)
 
c.setopt(c.HTTPPOST, [
    (‘fileupload‘, (
        # upload the contents of this file
        c.FORM_FILE, __file__,
        # specify a different file name for the upload
        c.FORM_FILENAME, ‘helloworld.py‘,
        # specify a different content type
        c.FORM_CONTENTTYPE, ‘application/x-python‘,
    )),
])
 
c.perform()
c.close()

import pycurl
 
c = pycurl.Curl()
c.setopt(c.URL, ‘http://pycurl.io/tests/testfileupload.php‘)
 
c.setopt(c.HTTPPOST, [
    (‘fileupload‘, (
        c.FORM_BUFFER, ‘readme.txt‘,
        c.FORM_BUFFERPTR, ‘This is a fancy readme file‘,
    )),
])
 
c.perform()
c.close()

import pycurl
 
c = pycurl.Curl()
c.setopt(c.URL, ‘ftp://ftp.sunet.se/‘)
c.setopt(c.FTP_USE_EPSV, 1)
c.setopt(c.QUOTE, [‘cwd pub‘, ‘type i‘])
c.perform()
c.close()

import pycurl
import threading
 
print >>sys.stderr, ‘Testing‘, pycurl.version
 
 
class Test(threading.Thread):
 
    def __init__(self, share):
        threading.Thread.__init__(self)
        self.curl = pycurl.Curl()
        self.curl.setopt(pycurl.URL, ‘http://curl.haxx.se‘)
        self.curl.setopt(pycurl.SHARE, share)
 
    def run(self):
        self.curl.perform()
        self.curl.close()
 
s = pycurl.CurlShare()
s.setopt(pycurl.SH_SHARE, pycurl.LOCK_DATA_COOKIE)
s.setopt(pycurl.SH_SHARE, pycurl.LOCK_DATA_DNS)
 
t1 = Test(s)
t2 = Test(s)
 
t1.start()
t2.start()
del s

import pycurl
 
m = pycurl.CurlMulti()
m.handles = []
c1 = pycurl.Curl()
c2 = pycurl.Curl()
c1.setopt(c1.URL, ‘http://curl.haxx.se‘)
c2.setopt(c2.URL, ‘http://cnn.com‘)
c2.setopt(c2.FOLLOWLOCATION, 1)
m.add_handle(c1)
m.add_handle(c2)
m.handles.append(c1)
m.handles.append(c2)
 
num_handles = len(m.handles)
while num_handles:
    while 1:
        ret, num_handles = m.perform()
        if ret != pycurl.E_CALL_MULTI_PERFORM:
            break
    m.select(1.0)
 
m.remove_handle(c2)
m.remove_handle(c1)
del m.handles
m.close()
c1.close()
c2.close()

import os, sys
try:
    from cStringIO import StringIO
except ImportError:
    from StringIO import StringIO
import pycurl
 
 
urls = (
    "http://curl.haxx.se",
    "http://www.python.org",
    "http://pycurl.sourceforge.net",
    "http://pycurl.sourceforge.net/tests/403_FORBIDDEN",  # that actually exists ;-)
    "http://pycurl.sourceforge.net/tests/404_NOT_FOUND",
)
 
# Read list of URIs from file specified on commandline
try:
    urls = open(sys.argv[1], "rb").readlines()
except IndexError:
    # No file was specified
    pass
 
# init
m = pycurl.CurlMulti()
m.handles = []
for url in urls:
    c = pycurl.Curl()
    # save info in standard Python attributes
    c.url = url.strip()
    c.body = StringIO()
    c.http_code = -1
    m.handles.append(c)
    # pycurl API calls
    c.setopt(c.URL, c.url)
    c.setopt(c.WRITEFUNCTION, c.body.write)
    c.setopt(c.FOLLOWLOCATION,True)
    m.add_handle(c)
 
# get data
num_handles = len(m.handles)
while num_handles:
     while 1:
         ret, num_handles = m.perform()
         print ret,num_handles
         if ret != pycurl.E_CALL_MULTI_PERFORM:
             break
     # currently no more I/O is pending, could do something in the meantime
     # (display a progress bar, etc.)
     m.select(1.0)
 
# close handles
for c in m.handles:
    # save info in standard Python attributes
    c.http_code = c.getinfo(c.HTTP_CODE)
    # pycurl API calls
    m.remove_handle(c)
    c.close()
m.close()
 
# print result
for c in m.handles:
    data = c.body.getvalue()
    if 0:
        print "**********", c.url, "**********"
        print data
    else:
        print "%-53s http_code %3d, %6d bytes" % (c.url, c.http_code, len(data))

import os, sys
try:
    from cStringIO import StringIO
except ImportError:
    from StringIO import StringIO
import pycurl
 
 
urls = (
    "http://curl.haxx.se",
    "http://www.python.org",
    "http://pycurl.sourceforge.net",
    "http://pycurl.sourceforge.net/THIS_HANDLE_IS_CLOSED",
)
 
# init
m = pycurl.CurlMulti()
m.handles = []
for url in urls:
    c = pycurl.Curl()
    # save info in standard Python attributes
    c.url = url
    c.body = StringIO()
    c.http_code = -1
    c.debug = 0
    m.handles.append(c)
    # pycurl API calls
    c.setopt(c.URL, c.url)
    c.setopt(c.WRITEFUNCTION, c.body.write)
    c.setopt(c.FOLLOWLOCATION,True)
    m.add_handle(c)
 
# debug - close a handle
if 1:
    c = m.handles[3]
    c.debug = 1
    c.close()
 
# get data
num_handles = len(m.handles)
while num_handles:
    while 1:
        ret, num_handles = m.perform()
        if ret != pycurl.E_CALL_MULTI_PERFORM:
            break
    # currently no more I/O is pending, could do something in the meantime
    # (display a progress bar, etc.)
    m.select(1.0)
 
# close handles
for c in m.handles:
    # save info in standard Python attributes
    try:
        c.http_code = c.getinfo(c.HTTP_CODE)
    except pycurl.error:
        # handle already closed - see debug above
        assert c.debug
        c.http_code = -1
    # pycurl API calls
    if 0:
        m.remove_handle(c)
        c.close()
    elif 0:
        # in the C API this is the wrong calling order, but pycurl
        # handles this automatically
        c.close()
        m.remove_handle(c)
    else:
        # actually, remove_handle is called automatically on close
        c.close()
m.close()
 
# print result
for c in m.handles:
    data = c.body.getvalue()
    if 0:
        print "**********", c.url, "**********"
    else:
        print "%-53s http_code %3d, %6d bytes" % (c.url, c.http_code, len(data)）

import os,sys
import pycurl
from StringIO import StringIO
 
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.WRITEFUNCTION,self.body.write)
       self._curl.setopt(pycurl.FOLLOWLOCATION,True)
       self._curl.setopt(pycurl.NOSIGNAL,1)
 
   def perform(self):
       self._curl.perform()
 
   def close(self):
       self.http_code=self._curl.getinfo(pycurl.HTTP_CODE)
       self._curl.close()
 
for url in urls:
   url=url.strip()
   if not url or url[0] == ‘#‘:
     continue
   c=Curl(url)
   c.perform()
   c.close()
   print  url, c.http_code

real    2m46.134s
user    0m0.134s
sys 0m0.185s