def request(self, endpoint, post=None):
buffer = BytesIO()
ch = pycurl.Curl()
ch.setopt(pycurl.URL, Constants.API_URL + endpoint)
ch.setopt(pycurl.USERAGENT, self.userAgent)
ch.setopt(pycurl.WRITEFUNCTION, buffer.write)
ch.setopt(pycurl.FOLLOWLOCATION, True)
ch.setopt(pycurl.HEADER, True)
ch.setopt(pycurl.VERBOSE, False)
ch.setopt(pycurl.COOKIEFILE, os.path.join(self.IGDataPath, self.username, self.username + "-cookies.dat"))
ch.setopt(pycurl.COOKIEJAR, os.path.join(self.IGDataPath, self.username, self.username + "-cookies.dat"))
if post is not None:
ch.setopt(pycurl.POST, True)
ch.setopt(pycurl.POSTFIELDS, post)
if self.proxy:
ch.setopt(pycurl.PROXY, self.proxyHost)
if self.proxyAuth:
ch.setopt(pycurl.PROXYUSERPWD, self.proxyAuth)
ch.perform()
resp = buffer.getvalue()
header_len = ch.getinfo(pycurl.HEADER_SIZE)
header = resp[0: header_len]
body = resp[header_len:]
ch.close()
if self.debug:
print("REQUEST: " + endpoint)
if post is not None:
if not isinstance(post, list):
print("DATA: " + str(post))
print("RESPONSE: " + body)
return [header, json_decode(body)]
python类VERBOSE的实例源码
def create_virtual_endpoints(self, data, category):
'''
Description:
create a virtual endpoints
category:
SENSOR
METER
GAUGE
ONOFF
LEVEL_CONTROL
'''
self.data = data
self.category = category
uri = "virtualEndpoints/?category=" + self.category
api_url = self.url + uri
c = pycurl.Curl()
c.setopt(pycurl.URL, api_url)
c.setopt(pycurl.HTTPHEADER, ['Accept: application/json','Content-Type: application/json','charset=UTF-8'])
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.POSTFIELDS, self.data)
c.setopt(pycurl.VERBOSE, 1)
c.perform()
def create_rooms(self, data):
'''
Description:
create a room
'''
self.data = data
uri = "rooms/"
api_url = self.url + uri
c = pycurl.Curl()
c.setopt(pycurl.URL, api_url)
c.setopt(pycurl.HTTPHEADER, ['Accept: application/json','Content-Type: application/json','charset=UTF-8'])
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.POSTFIELDS, self.data)
c.setopt(pycurl.VERBOSE, 1)
c.perform()
def put_attributes_config(self, data, uuid):
'''
Description:
modify an attribute
'''
self.data = data
self.uuid = uuid
uri = "attributes/" + self.uuid + "/config"
api_url = self.url + uri
c = pycurl.Curl()
c.setopt(pycurl.URL, api_url)
c.setopt(pycurl.HTTPHEADER, ['Accept: application/json','Content-Type: application/json','charset=UTF-8'])
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.POSTFIELDS, self.data)
c.setopt(pycurl.VERBOSE, 1)
c.perform()
def put_attributes(self, data, uuid):
'''
Description:
set attribute value with application/json content
'''
self.data = data
self.uuid = uuid
uri = "attributes/" + self.uuid + "/value"
api_url = self.url + uri
c = pycurl.Curl()
c.setopt(pycurl.URL, api_url)
c.setopt(pycurl.HTTPHEADER, ['Accept: application/json','Content-Type: application/json','charset=UTF-8'])
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(pycurl.CUSTOMREQUEST, "PUT")
c.setopt(pycurl.POSTFIELDS, self.data)
c.setopt(pycurl.VERBOSE, 1)
c.perform()
def initHandle(self):
""" sets common options to curl handle """
self.c.setopt(pycurl.FOLLOWLOCATION, 1)
self.c.setopt(pycurl.MAXREDIRS, 5)
self.c.setopt(pycurl.CONNECTTIMEOUT, 30)
self.c.setopt(pycurl.NOSIGNAL, 1)
self.c.setopt(pycurl.NOPROGRESS, 1)
if hasattr(pycurl, "AUTOREFERER"):
self.c.setopt(pycurl.AUTOREFERER, 1)
self.c.setopt(pycurl.SSL_VERIFYPEER, 0)
self.c.setopt(pycurl.LOW_SPEED_TIME, 30)
self.c.setopt(pycurl.LOW_SPEED_LIMIT, 5)
#self.c.setopt(pycurl.VERBOSE, 1)
self.c.setopt(pycurl.USERAGENT,
"Mozilla/5.0 (Windows NT 6.1; Win64; x64;en; rv:5.0) Gecko/20110619 Firefox/5.0")
if pycurl.version_info()[7]:
self.c.setopt(pycurl.ENCODING, "gzip, deflate")
self.c.setopt(pycurl.HTTPHEADER, ["Accept: */*",
"Accept-Language: en-US,en",
"Accept-Charset: ISO-8859-1,utf-8;q=0.7,*;q=0.7",
"Connection: keep-alive",
"Keep-Alive: 300",
"Expect:"])
def request(self, endpoint, headers=None, post=None, first=True):
buffer = BytesIO()
ch = pycurl.Curl()
ch.setopt(pycurl.URL, endpoint)
ch.setopt(pycurl.USERAGENT, self.userAgent)
ch.setopt(pycurl.WRITEFUNCTION, buffer.write)
ch.setopt(pycurl.FOLLOWLOCATION, True)
ch.setopt(pycurl.HEADER, True)
if headers:
ch.setopt(pycurl.HTTPHEADER, headers)
ch.setopt(pycurl.VERBOSE, self.debug)
ch.setopt(pycurl.SSL_VERIFYPEER, False)
ch.setopt(pycurl.SSL_VERIFYHOST, False)
ch.setopt(pycurl.COOKIEFILE, self.settingsPath + self.username + '-cookies.dat')
ch.setopt(pycurl.COOKIEJAR, self.settingsPath + self.username + '-cookies.dat')
if post:
import urllib
ch.setopt(pycurl.POST, len(post))
ch.setopt(pycurl.POSTFIELDS, urllib.urlencode(post))
ch.perform()
resp = buffer.getvalue()
header_len = ch.getinfo(pycurl.HEADER_SIZE)
header = resp[0: header_len]
body = resp[header_len:]
ch.close()
if self.debug:
import urllib
print("REQUEST: " + endpoint)
if post is not None:
if not isinstance(post, list):
print('DATA: ' + urllib.unquote_plus(json.dumps(post)))
print("RESPONSE: " + body + "\n")
return [header, json_decode(body)]
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def curl_get(self, url, refUrl=None):
buf = cStringIO.StringIO()
curl = pycurl.Curl()
curl.setopt(curl.URL, url)
curl.setopt(curl.WRITEFUNCTION, buf.write)
curl.setopt(pycurl.SSL_VERIFYPEER, 0)
#curl.setopt(pycurl.SSL_VERIFYHOST, 0)
#curl.setopt(pycurl.HEADERFUNCTION, self.headerCookie)
curl.setopt(pycurl.VERBOSE, 0)
curl.setopt(pycurl.USERAGENT, 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.11; rv:46.0) Gecko/20100101 Firefox/46.0')
#curl.setopt(pycurl.HTTPGET,1)
#curl.setopt(pycurl.COOKIE, Cookie)
#curl.setopt(pycurl.POSTFIELDS, 'j_username={ngnms_user}&j_password={ngnms_password}'.format(**self.ngnms_login))
curl.setopt(pycurl.COOKIEJAR, '/htdocs/logs/py_cookie.txt')
curl.setopt(pycurl.COOKIEFILE, '/htdocs/logs/py_cookie.txt')
if refUrl:
curl.setopt(pycurl.REFERER, refUrl)
#curl.setopt(c.CONNECTTIMEOUT, 5)
#curl.setopt(c.TIMEOUT, 8)
curl.perform()
backinfo = ''
if curl.getinfo(pycurl.RESPONSE_CODE) == 200:
backinfo = buf.getvalue()
curl.close()
return backinfo
def handle_request(self):
curl_handle = pycurl.Curl()
# set default options.
curl_handle.setopt(pycurl.URL, self.request_url)
curl_handle.setopt(pycurl.REFERER, self.request_url)
curl_handle.setopt(pycurl.USERAGENT, self.useragent)
curl_handle.setopt(pycurl.TIMEOUT, self.curlopts['TIMEOUT'])
curl_handle.setopt(pycurl.CONNECTTIMEOUT, self.curlopts['CONNECTTIMEOUT'])
curl_handle.setopt(pycurl.HEADER, True)
#curl_handle.setopt(pycurl.VERBOSE, 1)
curl_handle.setopt(pycurl.FOLLOWLOCATION, 1)
curl_handle.setopt(pycurl.MAXREDIRS, 5)
if(self.request_headers and len(self.request_headers) > 0):
tmplist = list()
for(key, value) in self.request_headers.items():
tmplist.append(key + ':' + value)
curl_handle.setopt(pycurl.HTTPHEADER, tmplist)
#??????POST
curl_handle.setopt(pycurl.HTTPPROXYTUNNEL, 1)
curl_handle.setopt(pycurl.POSTFIELDS, self.request_body)
response = StringIO.StringIO()
curl_handle.setopt(pycurl.WRITEFUNCTION, response.write)
try:
curl_handle.perform()
except pycurl.error as error:
raise ChannelException(error, 5)
self.response_code = curl_handle.getinfo(curl_handle.HTTP_CODE)
header_size = curl_handle.getinfo(curl_handle.HEADER_SIZE)
resp_str = response.getvalue()
self.response_headers = resp_str[0 : header_size]
self.response_body = resp_str[header_size : ]
response.close()
curl_handle.close()
curl_httpclient.py 文件源码
项目：My-Web-Server-Framework-With-Python2.7
作者: syjsu
项目源码
文件源码
阅读 16
收藏 0
点赞 0
评论 0
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _curl_create(max_simultaneous_connections=None):
curl = pycurl.Curl()
if logging.getLogger().isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, _curl_debug)
curl.setopt(pycurl.MAXCONNECTS, max_simultaneous_connections or 5)
return curl
def _curl_create(max_simultaneous_connections=None):
curl = pycurl.Curl()
if logging.getLogger().isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, _curl_debug)
curl.setopt(pycurl.MAXCONNECTS, max_simultaneous_connections or 5)
return curl
def _curl_create():
curl = pycurl.Curl()
if gen_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, _curl_debug)
return curl
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _curl_create(max_simultaneous_connections=None):
curl = pycurl.Curl()
if logging.getLogger().isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, _curl_debug)
curl.setopt(pycurl.MAXCONNECTS, max_simultaneous_connections or 5)
return curl
def _curl_create():
curl = pycurl.Curl()
if gen_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, _curl_debug)
return curl
def init_handle(self):
"""
Sets common options to curl handle.
"""
self.setopt(pycurl.FOLLOWLOCATION, 1)
self.setopt(pycurl.MAXREDIRS, 5)
self.setopt(pycurl.CONNECTTIMEOUT, 30)
self.setopt(pycurl.NOSIGNAL, 1)
self.setopt(pycurl.NOPROGRESS, 1)
if hasattr(pycurl, "AUTOREFERER"):
self.setopt(pycurl.AUTOREFERER, 1)
self.setopt(pycurl.SSL_VERIFYPEER, 0)
# Interval for low speed, detects connection loss, but can abort dl if
# hoster stalls the download
self.setopt(pycurl.LOW_SPEED_TIME, 45)
self.setopt(pycurl.LOW_SPEED_LIMIT, 5)
# do not save the cookies
self.setopt(pycurl.COOKIEFILE, '')
self.setopt(pycurl.COOKIEJAR, '')
# self.setopt(pycurl.VERBOSE, 1)
self.setopt(
pycurl.USERAGENT,
'Mozilla/5.0 (Windows NT 10.0; Win64; rv:53.0) '
'Gecko/20100101 Firefox/53.0')
if pycurl.version_info()[7]:
self.setopt(pycurl.ENCODING, 'gzip,deflate')
self.headers.update(
{'Accept': "*/*",
'Accept-Language': "en-US,en",
'Accept-Charset': "ISO-8859-1,utf-8;q=0.7,*;q=0.7",
'Connection': "keep-alive",
'Keep-Alive': "300",
'Expect': ""})
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def _create_handle(self):
c = pycurl.Curl()
# c.setopt(pycurl.VERBOSE, 1)
c.setopt(pycurl.CONNECTTIMEOUT, CONNECT_TIMEOUT)
c.setopt(pycurl.TIMEOUT, TIMEOUT)
c.setopt(pycurl.ENCODING, 'gzip, deflate')
return c
def __init__(self, data_file, verbose=False, **kwargs):
self.file_type = 'mzurl'
# strip off the final slash, if it exists
if data_file[-1] == '/':
data_file = data_file[:-1]
# Likewise, html or other madness.
if any([data_file.lower().endswith(x) for x in ['html', 'raw', 'wiff']]):
data_file = ".".join(data_file.split(".")[:-1])
self.data_file = data_file # actually a URL to a file
self.verbose = verbose
self._scans = None # cache of scan_info results for the whole file
# A string with the name and path of an appropriate temp file
# (varies by platform)
fd, self.cookie_file_name = tempfile.mkstemp(text=True)
os.close(fd)
# Handle to libcurl object
self.crl = pycurl.Curl()
# set some general options
self.crl.setopt(pycurl.COOKIEFILE, self.cookie_file_name)
self.crl.setopt(pycurl.COOKIEJAR, self.cookie_file_name)
self.crl.setopt(pycurl.FOLLOWLOCATION, True)
self.crl.setopt(pycurl.VERBOSE, verbose)
self.output = cStringIO.StringIO()
self.crl.setopt(pycurl.WRITEFUNCTION, self.output.write)
# how would you store an info file?
#if os.path.exists(data_file + '.mzi'):
#self._info_file = data_file + '.mzi'
#info_fh = open(self._info_file)
#self._info_scans = cPickle.load(info_fh)
#info_fh.close()
#else:
#self._info_file = None
def pideURL(url,cookie=False,cookie_name='cookie.txt', contador_curl = 0):
time.sleep(2)
print ("\n"+url+"\n")
c = pycurl.Curl()
if cookie:
c.setopt(pycurl.COOKIEJAR, 'cookies/'+cookie_name)
c.setopt(pycurl.COOKIEFILE, 'cookies/'+cookie_name)
c.setopt(pycurl.URL, url)
c.setopt(pycurl.CONNECTTIMEOUT, 15)
c.setopt(pycurl.TIMEOUT, 25)
c.setopt(pycurl.HTTPHEADER, ['Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8' ,'Accept-Language: en-US,en;q=0.5' ,'Connection: keep-alive' ,'Content-Type: application/x-www-form-urlencoded' ,'Host: services6.arcgis.com' ,'Origin: https://sig-ruv.maps.arcgis.com' ,'Referer: https://sig-ruv.maps.arcgis.com/apps/webappviewer/index.html?id=1e3873d1c01749929457c7a7b9315cda'])
#c.setopt(pycurl.VERBOSE, 1)
b = BytesIO()
BytesIO
c.setopt(pycurl.WRITEFUNCTION, b.write)
try:
c.perform()
return b.getvalue()
#print (response_string)
b.close()
except Exception as e:
#log ('Razon:',e)
response_string = None
if contador_curl<=10:
time.sleep(5)
pideURL(url,contador_curl+1)
else:
print ('Error: ',url)
print ('Error log: ',e)
def curl_common_init(buf):
handle = pycurl.Curl()
handle.setopt(pycurl.WRITEDATA, buf)
handle.setopt(pycurl.HEADERFUNCTION, curl_hdr)
handle.setopt(pycurl.DEBUGFUNCTION, curl_debug)
handle.setopt(pycurl.USERPWD, '{}:{}'.format(_g.conf._user,_g.conf._pass))
handle.setopt(pycurl.FOLLOWLOCATION, True)
# avoid FTP CWD for fastest directory transversal
handle.setopt(pycurl.FTP_FILEMETHOD, pycurl.FTPMETHOD_NOCWD)
# we always set this flag and let the logging module
# handle filtering.
handle.setopt(pycurl.VERBOSE, True)
# use ipv4 for VPNs
handle.setopt(pycurl.IPRESOLVE, pycurl.IPRESOLVE_V4)
handle.setopt(pycurl.USE_SSL, True)
handle.setopt(pycurl.SSL_VERIFYPEER, False)
# XXX
handle.setopt(pycurl.SSL_VERIFYHOST, 0)
return handle
def _curl_create(self):
curl = pycurl.Curl()
if curl_log.isEnabledFor(logging.DEBUG):
curl.setopt(pycurl.VERBOSE, 1)
curl.setopt(pycurl.DEBUGFUNCTION, self._curl_debug)
return curl
def set_verbosity(self, level):
"Set verbosity to 1 to see transactions."
self.set_option(pycurl.VERBOSE, level)
