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类COOKIEFILE的实例源码
def CurlPOST(url, data, cookie):
c = pycurl.Curl()
b = StringIO.StringIO()
c.setopt(pycurl.URL, url)
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.HTTPHEADER,['Content-Type: application/json'])
# c.setopt(pycurl.TIMEOUT, 10)
c.setopt(pycurl.WRITEFUNCTION, b.write)
c.setopt(pycurl.COOKIEFILE, cookie)
c.setopt(pycurl.COOKIEJAR, cookie)
c.setopt(pycurl.POSTFIELDS, data)
c.perform()
html = b.getvalue()
b.close()
c.close()
return html
def get_user_login(self, token):
'''
Description:
login
'''
self.token = token
uri = "user/login?username="+self.login+"&token="+self.token
api_url = self.url + uri
c = pycurl.Curl()
output = BytesIO()
c.setopt(c.URL, api_url)
### Read the cookie File
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(c.WRITEFUNCTION, output.write)
c.perform()
return json.loads(output.getvalue())
def get_all_virtual_endpoints(self):
'''
Description:
get all virtual endpoints
'''
uri = "virtualEndpoints"
api_url = self.url + uri
c = pycurl.Curl()
output_init = BytesIO()
c.setopt(c.URL, api_url)
### Create the cookie File
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(c.WRITEFUNCTION, output_init.write)
c.perform()
return json.loads(output_init.getvalue())
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 get_virtual_endpoints_config(self, uuid):
'''
Description:
get virtual endpoints config
'''
self.uuid = uuid
uri = "virtualEndpoints/"+self.uuid+"/config"
api_url = self.url + uri
c = pycurl.Curl()
output_init = BytesIO()
c.setopt(c.URL, api_url)
### Create the cookie File
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(c.WRITEFUNCTION, output_init.write)
c.perform()
return json.loads(output_init.getvalue())
def get_virtual_endpoint(self, uuid, network="false", device="false", clusterEndpoints="false", config="false", icons="true",
bindings="false", descriptor="false", room="false", info="false", full="false", attributes="false"):
'''
Description:
get virtual endpoint
'''
self.uuid = uuid
uri = ("virtualEndpoints/"+self.uuid+"?network="+ network +"&device=" + device + "&clusterEndpoints=" + clusterEndpoints + "&config=" + config + "&icons=" + icons +
"&type=false&bindings=" + bindings + "&descriptor=" + descriptor + "&room=" + room + "&info=" + info + "&full=" + full + "&attributes=" + attributes)
api_url = self.url + uri
c = pycurl.Curl()
output_init = BytesIO()
c.setopt(c.URL, api_url)
### Create the cookie File
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(c.WRITEFUNCTION, output_init.write)
c.perform()
return json.loads(output_init.getvalue())
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 save_and_synchronize(self, wait="false", timeout=30):
'''
Description:
synchronize Zipato with the Server
'''
self.wait = wait
self.timeout = timeout
uri = "box/saveAndSynchronize?wait=" + self.wait + "&timeout=" + str(self.timeout)
api_url = self.url + uri
c = pycurl.Curl()
output_init = BytesIO()
c.setopt(c.URL, api_url)
### Create the cookie File
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(c.WRITEFUNCTION, output_init.write)
c.perform()
c.close()
return json.loads(output_init.getvalue())
def synchronize(self, ifneeded="false", wait="false", timeout=30):
'''
Description:
synchronize Zipato with the Server
'''
self.ifneeded = ifneeded
self.wait = wait
self.timeout = timeout
uri = "box/synchronize?ifNeeded=" + self.ifneeded + "wait=" + self.wait + "&timeout=" + str(self.timeout)
api_url = self.url + uri
c = pycurl.Curl()
output_init = BytesIO()
c.setopt(c.URL, api_url)
### Create the cookie File
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(c.WRITEFUNCTION, output_init.write)
c.perform()
c.close()
return json.loads(output_init.getvalue())
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 get (url, user_agent=UA, referrer=None):
"""Make a GET request of the url using pycurl and return the data
(which is None if unsuccessful)"""
data = None
databuffer = StringIO()
curl = pycurl.Curl()
curl.setopt(pycurl.URL, url)
curl.setopt(pycurl.FOLLOWLOCATION, 1)
curl.setopt(pycurl.CONNECTTIMEOUT, 5)
curl.setopt(pycurl.TIMEOUT, 8)
curl.setopt(pycurl.WRITEFUNCTION, databuffer.write)
curl.setopt(pycurl.COOKIEFILE, '')
if user_agent:
curl.setopt(pycurl.USERAGENT, user_agent)
if referrer is not None:
curl.setopt(pycurl.REFERER, referrer)
try:
curl.perform()
data = databuffer.getvalue()
except Exception:
pass
curl.close()
return data
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 CurlGET(url, cookie):
c = pycurl.Curl()
b = StringIO.StringIO()
c.setopt(pycurl.URL, url)
# c.setopt(pycurl.TIMEOUT, 10)
# c.setopt(pycurl.POST, 1)
c.setopt(pycurl.WRITEFUNCTION, b.write)
c.setopt(pycurl.COOKIEFILE, cookie)
c.setopt(pycurl.COOKIEJAR, cookie)
c.perform()
html = b.getvalue()
b.close()
c.close()
return html
def get_download_link(fs_id):
"""
??????
:param fs_id:
:return:
"""
curl = pycurl.Curl()
curl.setopt(pycurl.USERAGENT, const.USER_AGENT)
curl.setopt(pycurl.REFERER, const.PAN_REFER_URL)
buffers = StringIO()
request_dict = {
'channel': 'chunlei',
'timestamp': '1473685224',
'fidlist': [fs_id],
'type': 'dlink',
'web': 1,
'clienttype': 0,
'bdstoken': 'e0e895bb3ef7b0cb70899ee66b74e809',
'sign': decode_sign(parse_sign2('d76e889b6aafd3087ac3bd56f4d4053a', '3545d271c5d07ba27355d39da0c62a4ee06d2d25'))
}
target_url = const.PAN_API_URL + 'download?' + urllib.urlencode(request_dict)
curl.setopt(pycurl.URL, target_url)
curl.setopt(pycurl.WRITEDATA, buffers)
curl.setopt(pycurl.COOKIEFILE, "cookie.txt")
curl.perform()
body = buffers.getvalue()
buffers.close()
curl.close()
data = json.loads(body)
if data['errno']:
return None
return data['dlink'][0]['dlink']
def setRequestContext(self, url, get, post, referer, cookies, multipart=False):
""" sets everything needed for the request """
url = myquote(url)
if get:
get = urlencode(get)
url = "%s?%s" % (url, get)
self.c.setopt(pycurl.URL, url)
self.c.lastUrl = url
if post:
self.c.setopt(pycurl.POST, 1)
if not multipart:
if type(post) == unicode:
post = str(post) #unicode not allowed
elif type(post) == str:
pass
else:
post = myurlencode(post)
self.c.setopt(pycurl.POSTFIELDS, post)
else:
post = [(x, y.encode('utf8') if type(y) == unicode else y ) for x, y in post.iteritems()]
self.c.setopt(pycurl.HTTPPOST, post)
else:
self.c.setopt(pycurl.POST, 0)
if referer and self.lastURL:
self.c.setopt(pycurl.REFERER, str(self.lastURL))
if cookies:
self.c.setopt(pycurl.COOKIEFILE, "")
self.c.setopt(pycurl.COOKIEJAR, "")
self.getCookies()
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 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 __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 pideURL(self,url,compressed = False, cookie=False, contador_curl = 0):
time.sleep(3)
Scrape.contador+=1
print ("\n"+url)
print ("\n\t.l."+str(Scrape.contador))
c = pycurl.Curl()
if cookie:
c.setopt(pycurl.COOKIEJAR, 'cookie.txt')
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(pycurl.URL, url)
c.setopt(pycurl.CONNECTTIMEOUT, 15)
c.setopt(pycurl.TIMEOUT, 25)
c.setopt(pycurl.HTTPHEADER, self.headers)
c.setopt( pycurl.PROXY, '127.0.0.1' )
c.setopt( pycurl.PROXYPORT, 9050 )
c.setopt( pycurl.PROXYTYPE, pycurl.PROXYTYPE_SOCKS5_HOSTNAME )
b = BytesIO()
BytesIO
c.setopt(pycurl.WRITEFUNCTION, b.write)
self.url = url
try:
c.perform()
self.response_string = b.getvalue()
#print (self.response_string)
b.close()
except Exception as e:
#self.log ('Razon:',e)
self.response_string = None
if contador_curl<=10:
time.sleep(5)
self.pideURL(url,contador_curl+1)
else:
print ('Error: ',url)
print ('Error log: ',e)
def pidePOST(self,url,data,compressed = False,cookie=False, contador_curl = 0, debug=False):
time.sleep(3)
Scrape.contador+=1
print ("\n"+url)
print ("\n\t.l."+str(Scrape.contador))
c = pycurl.Curl()
if cookie:
c.setopt(pycurl.COOKIEJAR, 'cookie.txt')
c.setopt(pycurl.COOKIEFILE, 'cookie.txt')
c.setopt(pycurl.URL, url)
c.setopt(pycurl.CONNECTTIMEOUT, 15)
c.setopt(pycurl.TIMEOUT, 25)
c.setopt(pycurl.HTTPHEADER, self.headers)
if compressed:
c.setopt(pycurl.ENCODING, 'gzip,deflate')
c.setopt(c.POSTFIELDS, data)
if debug:
c.setopt(c.VERBOSE, True)
c.setopt( pycurl.PROXY, '127.0.0.1' )
c.setopt( pycurl.PROXYPORT, 9050 )
c.setopt( pycurl.PROXYTYPE, pycurl.PROXYTYPE_SOCKS5_HOSTNAME )
b = BytesIO()
BytesIO
c.setopt(pycurl.WRITEFUNCTION, b.write)
self.url = url
try:
c.perform()
self.response_string = b.getvalue()
#print (self.response_string)
b.close()
except Exception as e:
#print ('Razon:',e)
self.response_string = None
def __init__(self, base_url="", fakeheaders=[]):
self.handle = pycurl.Curl()
# These members might be set.
self.set_url(base_url)
self.verbosity = 0
self.fakeheaders = fakeheaders
# Nothing past here should be modified by the caller.
self.payload = None
self.payload_io = BytesIO()
self.hrd = ""
# Verify that we've got the right site; harmless on a non-SSL connect.
self.set_option(pycurl.SSL_VERIFYHOST, 2)
# Follow redirects in case it wants to take us to a CGI...
self.set_option(pycurl.FOLLOWLOCATION, 1)
self.set_option(pycurl.MAXREDIRS, 5)
self.set_option(pycurl.NOSIGNAL, 1)
# Setting this option with even a nonexistent file makes libcurl
# handle cookie capture and playback automatically.
self.set_option(pycurl.COOKIEFILE, "/dev/null")
# Set timeouts to avoid hanging too long
self.set_timeout(30)
# Use password identification from .netrc automatically
self.set_option(pycurl.NETRC, 1)
self.set_option(pycurl.WRITEFUNCTION, self.payload_io.write)
def header_callback(x):
self.hdr += x.decode('ascii')
self.set_option(pycurl.HEADERFUNCTION, header_callback)
def list_dir(dir_name):
"""
????????
:param dir_name: ??
:return:
"""
result = list()
curl = pycurl.Curl()
curl.setopt(pycurl.USERAGENT, const.USER_AGENT)
curl.setopt(pycurl.REFERER, const.PAN_REFER_URL)
buffers = StringIO()
request_dict = {
'channel': 'chunlei',
'clienttype': 0,
'showempty': 0,
'web': 1,
'order': 'time',
'desc': 1,
'page': 1,
'num': 100,
'dir': dir_name,
'bdstoken': 'e0e895bb3ef7b0cb70899ee66b74e809'
}
target_url = const.PAN_API_URL + 'list?' + urllib.urlencode(request_dict)
curl.setopt(pycurl.URL, target_url)
curl.setopt(pycurl.WRITEDATA, buffers)
curl.setopt(pycurl.COOKIEFILE, "cookie.txt")
curl.perform()
body = buffers.getvalue()
print body
buffers.close()
curl.close()
data = json.loads(body)
if data['errno'] == 0:
for a_list in data['list']:
dlink = get_download_link(a_list['fs_id'])
if dlink:
dlink = dlink.replace('\\', '')
result.append(dlink)
return result
