def test_post(self):
curl = CurlStub(b"result")
result = fetch("http://example.com", post=True, curl=curl)
self.assertEqual(result, b"result")
self.assertEqual(curl.options,
{pycurl.URL: b"http://example.com",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.POST: True,
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
python类MAXREDIRS的实例源码
def test_post_data(self):
curl = CurlStub(b"result")
result = fetch("http://example.com", post=True, data="data", curl=curl)
self.assertEqual(result, b"result")
self.assertEqual(curl.options[pycurl.READFUNCTION](), b"data")
self.assertEqual(curl.options,
{pycurl.URL: b"http://example.com",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.POST: True,
pycurl.POSTFIELDSIZE: 4,
pycurl.READFUNCTION: Any(),
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
def test_cainfo(self):
curl = CurlStub(b"result")
result = fetch("https://example.com", cainfo="cainfo", curl=curl)
self.assertEqual(result, b"result")
self.assertEqual(curl.options,
{pycurl.URL: b"https://example.com",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.CAINFO: b"cainfo",
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
def test_headers(self):
curl = CurlStub(b"result")
result = fetch("http://example.com",
headers={"a": "1", "b": "2"}, curl=curl)
self.assertEqual(result, b"result")
self.assertEqual(curl.options,
{pycurl.URL: b"http://example.com",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.HTTPHEADER: ["a: 1", "b: 2"],
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
def test_pycurl_insecure(self):
curl = CurlStub(b"result")
result = fetch("http://example.com/get-ca-cert", curl=curl,
insecure=True)
self.assertEqual(result, b"result")
self.assertEqual(curl.options,
{pycurl.URL: b"http://example.com/get-ca-cert",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.SSL_VERIFYPEER: False,
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
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 download_preparing(self, pyfile):
if not isinstance(pyfile.plugin.req, HTTPRequest):
return
connecttimeout = self.config.get('connecttimeout')
maxredirs = self.config.get('maxredirs')
useragent = self.config.get('useragent')
if connecttimeout:
pyfile.plugin.req.http.c.setopt(
pycurl.CONNECTTIMEOUT, connecttimeout)
if maxredirs:
pyfile.plugin.req.http.c.setopt(pycurl.MAXREDIRS, maxredirs)
if useragent:
self.log_debug("Use custom user-agent string `%s`" % useragent)
pyfile.plugin.req.http.c.setopt(
pycurl.USERAGENT, encode(useragent))
def Curl(url,headers):
while 1:
try:
c = pycurl.Curl()
c.setopt(pycurl.REFERER, 'http://weixin.sogou.com/')
c.setopt(pycurl.FOLLOWLOCATION, True)
c.setopt(pycurl.MAXREDIRS,5)
c.setopt(pycurl.CONNECTTIMEOUT, 60)
c.setopt(pycurl.TIMEOUT,120)
c.setopt(pycurl.ENCODING, 'gzip,deflate')
c.fp = StringIO.StringIO()
c.setopt(pycurl.URL, url)
c.setopt(pycurl.HTTPHEADER,headers)
c.setopt(c.WRITEFUNCTION, c.fp.write)
c.perform()
html = c.fp.getvalue()
if '??????' in html:
print u'??????,??10??'
time.sleep(600)
else:
return html
except Exception, e:
print url,'curl(url)',e
continue
#????????
def getHtml(url,headers):
c = pycurl.Curl() #??curl????????
c.setopt(pycurl.URL, url) #??????URL
c.setopt(pycurl.FOLLOWLOCATION, True) #????????
c.setopt(pycurl.MAXREDIRS,5) #?????????
c.setopt(pycurl.CONNECTTIMEOUT, 60) #??????
c.setopt(pycurl.TIMEOUT,120) #????
c.setopt(pycurl.ENCODING, 'gzip,deflate') #??gzip???????????????????gzip?????????gzip??????
c.fp = StringIO.StringIO() #??StringIO??
c.setopt(pycurl.HTTPHEADER,headers) #?????
c.setopt(pycurl.POST, 1) #??get
c.setopt(pycurl.POSTFIELDS, data) #??POST??
c.setopt(c.WRITEFUNCTION, c.fp.write) #???????
c.perform() #??
html = c.fp.getvalue() #?????
return html
def download_preparing(self, pyfile):
if not isinstance(pyfile.plugin.req, HTTPRequest):
return
connecttimeout = self.config.get('connecttimeout')
maxredirs = self.config.get('maxredirs')
useragent = self.config.get('useragent')
if connecttimeout:
pyfile.plugin.req.http.c.setopt(
pycurl.CONNECTTIMEOUT, connecttimeout)
if maxredirs:
pyfile.plugin.req.http.c.setopt(pycurl.MAXREDIRS, maxredirs)
if useragent:
self.log_debug("Use custom user-agent string `%s`" % useragent)
pyfile.plugin.req.http.c.setopt(
pycurl.USERAGENT, encode(useragent))
def test_basic(self):
curl = CurlStub(b"result")
result = fetch("http://example.com", curl=curl)
self.assertEqual(result, b"result")
self.assertEqual(curl.options,
{pycurl.URL: b"http://example.com",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
def test_create_curl(self):
curls = []
def pycurl_Curl():
curl = CurlStub(b"result")
curls.append(curl)
return curl
Curl = pycurl.Curl
try:
pycurl.Curl = pycurl_Curl
result = fetch("http://example.com")
curl = curls[0]
self.assertEqual(result, b"result")
self.assertEqual(curl.options,
{pycurl.URL: b"http://example.com",
pycurl.FOLLOWLOCATION: 1,
pycurl.MAXREDIRS: 5,
pycurl.CONNECTTIMEOUT: 30,
pycurl.LOW_SPEED_LIMIT: 1,
pycurl.LOW_SPEED_TIME: 600,
pycurl.NOSIGNAL: 1,
pycurl.WRITEFUNCTION: Any(),
pycurl.DNS_CACHE_TIMEOUT: 0,
pycurl.ENCODING: b"gzip,deflate"})
finally:
pycurl.Curl = Curl
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()
def _login(self):
try:
c = pycurl.Curl()
c.setopt(pycurl.CAINFO, certifi.where())
c.setopt(pycurl.URL, self.url)
b = StringIO.StringIO()
c.setopt(pycurl.USERAGENT, "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 1.1.4322)")
c.setopt(pycurl.WRITEFUNCTION, b.write)
c.setopt(pycurl.FOLLOWLOCATION, 1)
c.setopt(pycurl.MAXREDIRS, 5)
c.setopt(pycurl.CUSTOMREQUEST, "POST")
c.setopt(pycurl.POSTFIELDS, self.post_data)
c.perform()
if b.getvalue():
logging.info('success login') # For INFO level
self.API_TOKEN = json.loads(b.getvalue())["access_token"]
self.save_token()
else:
logging.warning('success fail,get null result') #2 For WARNING level
logging.debug(self.API_TOKEN)
b.close()
c.close()
except pycurl.E_HTTP_POST_ERROR:
logging.error(str(pycurl.E_HTTP_POST_ERROR))
except Exception as e:
logging.error('please check your password or username')
logging.error(e.message) #3 For ERROR level
pass
def getKeyword(i):#??json
try:
time.sleep(1)
headers = [
'Host:fengchao.baidu.com',
'User-Agent: %s' %getUA(),
'Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language: zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3',
'Accept-Encoding: gzip, deflate',
'Referer: http://fengchao.baidu.com/nirvana/main.html?userid=8048066',
'Connection: keep-alive',
'COOKIE:%s' %COOKIE,
]
post = urllib.urlencode({
'params': '{"entry":"kr_station","query":"%s","querytype":1,"pageNo":1,"pageSize":300}' % keyword_list[i],
'path': 'jupiter/GET/kr/word',
'token': TOKEN,
'userid': USERID,
})
url = 'http://fengchao.baidu.com/nirvana/request.ajax?path=jupiter/GET/kr/word'
c = pycurl.Curl()
# c.setopt(pycurl.PROXY, getRandomAlbIp())
c.setopt(pycurl.URL, url)
c.setopt(pycurl.FOLLOWLOCATION, True)
c.setopt(pycurl.MAXREDIRS,5)
c.setopt(pycurl.CONNECTTIMEOUT, 20)
c.setopt(pycurl.TIMEOUT,20)
c.setopt(pycurl.ENCODING, 'gzip,deflate')
c.fp = StringIO.StringIO()
c.setopt(pycurl.HTTPHEADER,headers)
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.POSTFIELDS, post)
c.setopt(c.WRITEFUNCTION, c.fp.write)
c.perform()
# mutex.acquire()#??
jsonData = c.fp.getvalue()
analyseJsonData(i,jsonData)
# mutex.release()#??
except Exception,e:
print e
pass
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, 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 to_pycurl_object(c, req):
c.setopt(pycurl.MAXREDIRS, 5)
c.setopt(pycurl.WRITEFUNCTION, req.body_callback)
c.setopt(pycurl.HEADERFUNCTION, req.header_callback)
c.setopt(pycurl.NOSIGNAL, 1)
c.setopt(pycurl.SSL_VERIFYPEER, False)
c.setopt(pycurl.SSL_VERIFYHOST, 0)
c.setopt(pycurl.URL,req.completeUrl)
if req.getConnTimeout():
c.setopt(pycurl.CONNECTTIMEOUT, req.getConnTimeout())
if req.getTotalTimeout():
c.setopt(pycurl.TIMEOUT, req.getTotalTimeout())
authMethod, userpass = req.getAuth()
if authMethod or userpass:
if authMethod == "basic":
c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
elif authMethod == "ntlm":
c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_NTLM)
elif authMethod == "digest":
c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_DIGEST)
c.setopt(pycurl.USERPWD, userpass)
c.setopt(pycurl.HTTPHEADER, req.getHeaders())
if req.method == "POST":
c.setopt(pycurl.POSTFIELDS, req.postdata)
if req.method != "GET" and req.method != "POST":
c.setopt(pycurl.CUSTOMREQUEST, req.method)
if req.method == "HEAD":
c.setopt(pycurl.NOBODY, True)
if req.followLocation:
c.setopt(pycurl.FOLLOWLOCATION, 1)
proxy = req.getProxy()
if proxy != None:
c.setopt(pycurl.PROXY, proxy)
if req.proxytype=="SOCKS5":
c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS5)
elif req.proxytype=="SOCKS4":
c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS4)
req.delHeader("Proxy-Connection")
return c
def getOneGameODD(game):
resultStr = ''
try:
oddURL = 'http://27.45.161.37:8072/phone/1x2.aspx?ID=' + str(game.soccerID) + '&an=iosQiuTan&av=5.9&from=2&lang=0&subversion=1'
# print oddURL
except:
pass
c = pycurl.Curl()
c.setopt(pycurl.URL, oddURL)
b = StringIO.StringIO()
c.setopt(pycurl.WRITEFUNCTION, b.write)
c.setopt(pycurl.FOLLOWLOCATION, 1)
c.setopt(pycurl.MAXREDIRS, 5)
c.perform()
resultStr = b.getvalue().decode('utf8')
if resultStr != '':
array = resultStr.split('!')
companys = []
for unit in array:
# print unit.decode('utf-8')
company = BetCompany()
company.league = game.leauge;
company.result = game.soccer
company.homeSoccer = game.allHome
company.friendSoccer = game.allFriend
company.soccerGameId = game.soccerID
unitArray = unit.split('^')
try:
company.companyTitle = unitArray[0].encode('utf-8')
company.orignal_winOdd = float(unitArray[2])
company.orignal_drawOdd = float(unitArray[3])
company.orignal_loseOdd = float(unitArray[4])
company.winOdd = float(unitArray[5])
company.drawOdd = float(unitArray[6])
company.loseOdd = float(unitArray[7])
except IndexError as e:
print e
print unitArray
if company.companyTitle in ['????', '10BET', 'bet 365', 'bwin', 'Interwetten', 'SB', '??', '??', '????', '????', '??','Oddset','SNAI','ManbetX']:
companys.append(company)
if company.companyTitle == '??':
game.orignal_aomenOdd = (company.orignal_winOdd, company.orignal_drawOdd, company.orignal_loseOdd)
game.now_aomenOdd = (company.winOdd, company.drawOdd, company.loseOdd)
return companys
else:
return []
def to_pycurl_object(c, req):
c.setopt(pycurl.MAXREDIRS, 5)
c.setopt(pycurl.WRITEFUNCTION, req.body_callback)
c.setopt(pycurl.HEADERFUNCTION, req.header_callback)
c.setopt(pycurl.NOSIGNAL, 1)
c.setopt(pycurl.SSL_VERIFYPEER, False)
c.setopt(pycurl.SSL_VERIFYHOST, 0)
c.setopt(pycurl.URL,req.completeUrl)
if req.getConnTimeout():
c.setopt(pycurl.CONNECTTIMEOUT, req.getConnTimeout())
if req.getTotalTimeout():
c.setopt(pycurl.TIMEOUT, req.getTotalTimeout())
authMethod, userpass = req.getAuth()
if authMethod or userpass:
if authMethod == "basic":
c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
elif authMethod == "ntlm":
c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_NTLM)
elif authMethod == "digest":
c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_DIGEST)
c.setopt(pycurl.USERPWD, userpass)
c.setopt(pycurl.HTTPHEADER, req.getHeaders())
if req.method == "POST":
c.setopt(pycurl.POSTFIELDS, req.postdata)
if req.method != "GET" and req.method != "POST":
c.setopt(pycurl.CUSTOMREQUEST, req.method)
if req.method == "HEAD":
c.setopt(pycurl.NOBODY, True)
if req.followLocation:
c.setopt(pycurl.FOLLOWLOCATION, 1)
proxy = req.getProxy()
if proxy != None:
c.setopt(pycurl.PROXY, proxy)
if req.proxytype=="SOCKS5":
c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS5)
elif req.proxytype=="SOCKS4":
c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS4)
req.delHeader("Proxy-Connection")
return c
