Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
opts.verbose = opts.verbose - 1
print "Python", sys.version
print "PycURL %s (compiled against 0x%x)" % (pycurl.version, pycurl.COMPILE_LIBCURL_VERSION_NUM)
print "PycURL version info", pycurl.version_info()
print " %s, compiled %s" % (pycurl.__file__, pycurl.COMPILE_DATE)
# /***********************************************************************
# // test misc
# ************************************************************************/
if 1:
c = Curl()
assert c.URL is pycurl.URL
del c
# /***********************************************************************
# // test handles
# ************************************************************************/
# remove an invalid handle: this should fail
if 1:
m = CurlMulti()
c = Curl()
try:
m.remove_handle(c)
except pycurl.error:
pass
else:
def sendMultipleMetricsThread(self,url,payload,count):
for i in range(0,count):
for j in range(0,self.errorRetry):
cbuffer = StringIO()
c = pycurl.Curl()
c.setopt(pycurl.URL,url )
c.setopt(pycurl.HTTPHEADER, ['X-Postmark-Server-Token: API_TOKEN_HERE','Accept: application/json'])
c.setopt(pycurl.POST, 1)
c.setopt(c.WRITEDATA, cbuffer)
c.setopt(pycurl.POSTFIELDS, payload)
c.perform()
# if len(cbuffer.getvalue()) >0: print buffer.getvalue()
c.close()
if cbuffer.getvalue().find("error") != -1:
# logging.info("sendMultipleMetricsThread: Error Retry "+j)
if j >= self.errorRetry - 1:
logging.info("sendMultipleMetricsThread: Max Error Retry Failure")
time.sleep(1)
continue
break
def sendMultipleMetrics(self,node,tsname,count):
"""Do the actual annotation work"""
#parse doc to see if annotations already present
with open(infile,"rb") as f:
self.doc = minidom.parse(f)
if len(self.doc.getElementsByTagName("annotationART")) > 0:
self.__upgradeXML()
elif len(self.doc.getElementsByTagName("CoreSc1")) < 1:
pdata = [('paper', (pycurl.FORM_FILE, infile) )]
c = pycurl.Curl()
c.setopt(pycurl.URL, SAPIENTA_URL)
c.setopt(pycurl.POST,1)
c.setopt(pycurl.HTTPPOST, pdata)
logging.info("Uploading %s to annotation server", infile)
self.perform(c)
tmpnam, sents = self.result.split(":")
labels = sents.split(">")
self.__annotateXML( labels )
with codecs.open(outfile,'w', encoding='utf-8') as f:
self.doc.writexml(f)
tf.write("output console=off\n")
tf.write("output script=off\n")
tf.write("output error=merge\n")
tf.write("set limit 1\n")
tf.write("format object fmt1 \"%IDLIST(HD|HIP|1) | %OTYPELIST(S) | %SP(S)\"\n")
tf.write("result full\n")
tf.write("query sample region(circle, %s %s,5s) & otype='Star'\n" % (h[0].header['HIERARCH ESO TEL TARG ALPHA'],h[0].header['HIERARCH ESO TEL TARG DELTA'] ))
#tf.write("set radius 5s\n")
tf.close()
values = [
("scriptFIle", (pycurl.FORM_FILE, tfile))
]
output = StringIO.StringIO()
c = pycurl.Curl()
c.setopt(pycurl.URL, "http://simbad.harvard.edu/simbad/sim-script")
c.setopt(c.HTTPPOST, values)
c.setopt(pycurl.WRITEFUNCTION, output.write)
c.perform()
c.close()
result = output.getvalue()
lines = result.split('\n')
result = lines[len(lines)-3]
if (result.count('No') > 0):
# build alternate obname based on ra-dec
ra_s = h[0].header['HIERARCH ESO OBS ALPHACAT'].replace('h','').replace('m','')
dec_s = h[0].header['HIERARCH ESO OBS DELTACAT'].replace(':','')
hour_l = len( h[0].header['HIERARCH ESO OBS ALPHACAT'].split('h')[0] )
if (hour_l == 1):
obname = 'J_0'+ra_s+dec_s
elif (hour_l == 2):
crl.fp = StringIO.StringIO()
if method == "GET" and data:
self.url = "%s?%s" % (self.url, urlencode(data))
elif method == "POST" and data:
crl.setopt(pycurl.POSTFIELDS, urlencode(data)) # post data
elif method == "UPLOAD" and data:
if isinstance(data, dict):
upload_data = data.items()
else:
upload_data = data
crl.setopt(pycurl.HTTPPOST, upload_data) # upload file
crl.setopt(pycurl.URL, self.url)
crl.setopt(crl.WRITEFUNCTION, crl.fp.write)
try:
crl.perform()
except Exception:
return None
crl.close()
back = crl.fp.getvalue()
crl.fp.close()
return back
"with known CA certificates.")
raise TransferException("Peer certificate could not be "
"authenticated with known CA "
"certificates.")
else:
log_error(e[1])
raise
status = curl.getinfo(pycurl.HTTP_CODE)
if status == 200 or (URL(url).scheme == "file" and status == 0):
# OK or file
break
elif status in (301, 302): # redirects
url = curl.getinfo(pycurl.REDIRECT_URL)
log_debug(2, "Got redirect to %s" % url)
curl.setopt(pycurl.URL, url)
else:
log_error("Connecting to %s has failed after %s "
"tries with HTTP error code %s." %
(URL(url).getURL(stripPw=True), connect_retries, status))
raise TransferException("Connection failed after %s tries with "
"HTTP error %s." % (connect_retries, status))
# StringIO.write leaves the cursor at the end of the file
response.seek(0)
return response
upload_url = str(url_candidate['url'])
if not upload_url:
raise Exception("Unable to extract disk upload URL from HttpNfcLease")
self.log.debug("Extracted image upload URL (%s) from lease" % (upload_url))
lease_timeout = lease.info.leaseTimeout
self.time_at_last_poke = time()
image_file = open(imagefilename)
# Upload the image itself
image_size = os.path.getsize(imagefilename)
curl = pycurl.Curl()
curl.setopt(pycurl.URL, upload_url)
curl.setopt(pycurl.SSL_VERIFYPEER, 0)
curl.setopt(pycurl.POST, 1)
curl.setopt(pycurl.POSTFIELDSIZE, image_size)
curl.setopt(pycurl.READFUNCTION, image_file.read)
curl.setopt(pycurl.HTTPHEADER, ["User-Agent: Load Tool (PyCURL Load Tool)", "Content-Type: application/octet-stream"])
curl.setopt(pycurl.NOPROGRESS, 0)
curl.setopt(pycurl.PROGRESSFUNCTION, self.curl_progress)
curl.perform()
curl.close()
image_file.close()
lease.HttpNfcLeaseComplete()
vm = lease.info.entity
def create_job(self, title, instructions, cml=None, callback=None):
url = "http://api.crowdflower.com/v1/jobs.json"
post_dict = {'key':self.key,
'job[title]':title,
'job[instructions]':instructions}
if cml: post_dict['job[problem]'] = cml
post_fields = urllib.urlencode(post_dict)
c = pycurl.Curl()
c.setopt(pycurl.POST, True)
c.setopt(pycurl.URL, url)
c.setopt(pycurl.POSTFIELDS, post_fields)
if callback: c.setopt(pycurl.WRITEFUNCTION, callback)
c.setopt(pycurl.HTTPHEADER, ['Content-Type : application/json', ])
c.perform()
c.close()
def dxDown(url, fullpath):
c=pycurl.Curl() # 縮寫一下
c.setopt(c.FOLLOWLOCATION, True) # 允許重定向
c.setopt(pycurl.USERAGENT, b"Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 1.1.4322)") # 模擬瀏覽器
c.setopt(pycurl.URL, url) # 訪問指定網址
c.setopt(pycurl.COOKIEJAR, 'cookie.txt') # 把 cookie 存到文件
c.setopt(pycurl.COOKIEFILE, "cookie.txt") # 用文件掛 cookie
f = open(fullpath, 'wb') # 定義一個文件
c.setopt(c.WRITEDATA, f) # 指定返回信息的寫入文件,或作 c.setopt(c.WRITEFUNCTION, f.write)
c.perform() # 獲得服務器返回信息
f.close()
if c.getinfo(pycurl.HTTP_CODE) != 200:
os.remove(fullpath)
print("Failed!")
contents.sort()
parentdir = os.path.normpath(os.path.join(path, '..'))
b.write(directory_listing % (path, path, 'file://' + parentdir))
for f in contents:
target = 'file://%s/%s' % (path, f)
if os.path.isdir(os.path.join(path, f)):
b.write('%s<a href="%s">%s</a><br>' % (I_DIRECTORY, target, f))
else:
b.write('%s<a href="%s">%s</a><br>' % (I_REGULAR, target, f))
b.write('<hr>')
self.render.append((b, handle))
continue
else:
url = 'file://%s' % path
curl.setopt(pycurl.WRITEFUNCTION, b.write)
curl.setopt(pycurl.URL, url)
try:
curl.perform()
except pycurl.error, msg:
b.write(internal_error % msg[1])
except:
msg = "Error retrieving URL: %s" % url
b.write(internal_error % msg)
# Flag empty documents to the renderer
if b.tell() == 0:
b.close()
b = None
# Enqueue the document on the rendering pipeline
self.render.append((b, handle))