def download(self, url, retry_count=3, headers=None, proxy=None, data=None):
if url is None:
return None
try:
req = request.Request(url, headers=headers, data=data)
cookie = cookiejar.CookieJar()
cookie_process = request.HTTPCookieProcessor(cookie)
opener = request.build_opener()
if proxy:
proxies = {urlparse(url).scheme: proxy}
opener.add_handler(request.ProxyHandler(proxies))
content = opener.open(req).read()
except error.URLError as e:
print('HtmlDownLoader download error:', e.reason)
content = None
if retry_count > 0:
if hasattr(e, 'code') and 500 <= e.code < 600:
#??? HTTPError ??? HTTP CODE ? 5XX ???????????????????
return self.download(url, retry_count-1, headers, proxy, data)
return content
python类CookieJar()的实例源码
def __init__(self, ssl_context, debug=False, proxies=None, no_proxy=None,
cookie=None, http_basicauth=None, headers=None):
"""
@param ssl_context: SSL context to use with this configuration
@type ssl_context: OpenSSL.SSL.Context
@param debug: if True, output debugging information
@type debug: bool
@param proxies: proxies to use for
@type proxies: dict with basestring keys and values
@param no_proxy: hosts for which a proxy should not be used
@type no_proxy: basestring
@param cookie: cookies to set for request
@type cookie: cookielib.CookieJar (python 3 - http.cookiejar)
@param http_basicauth: http authentication, or None
@type http_basicauth: tuple of (username,password)
@param headers: http headers
@type headers: dict
"""
self.ssl_context = ssl_context
self.debug = debug
self.proxies = proxies
self.no_proxy = no_proxy
self.cookie = cookie
self.http_basicauth = http_basicauth
self.headers = headers
def logout(self):
'''
????
:returns: True????False????????
'''
#passport_logout_response = self.get_response(logout_url)
self.session.cookies = cookielib.CookieJar()
response = self.get_response(logout_url)
check_logout = re.findall('????', response)
if len(check_logout) > 0:
self.logined = False
self.remove_cookies()
return True
else:
return False
def __init__(self, timeout=None, proxy=None, cacert=None, sessions=False):
if (timeout is not None) and not self.supports_feature('timeout'):
raise RuntimeError('timeout is not supported with urllib2 transport')
if proxy:
raise RuntimeError('proxy is not supported with urllib2 transport')
if cacert:
raise RuntimeError('cacert is not support with urllib2 transport')
handlers = []
if ((sys.version_info[0] == 2 and sys.version_info >= (2,7,9)) or
(sys.version_info[0] == 3 and sys.version_info >= (3,2,0))):
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
handlers.append(urllib2.HTTPSHandler(context=context))
if sessions:
handlers.append(urllib2.HTTPCookieProcessor(CookieJar()))
opener = urllib2.build_opener(*handlers)
self.request_opener = opener.open
self._timeout = timeout
def cookie_friendly_download(referer_url, file_url, store_dir='.', timeout=1000):
from http.cookiejar import CookieJar
from urllib import request
cj = CookieJar()
cp = request.HTTPCookieProcessor(cj)
opener = request.build_opener(cp)
with opener.open(referer_url) as fin:
fin.headers.items()
import os
from os import path
with opener.open(file_url, timeout=timeout) as fin:
file_bin = fin.read()
filename = fin.headers['Content-Disposition']
filename = filename.split(';')[-1].split('=')[1]
os.makedirs(store_dir, exist_ok=True)
with open(path.join(store_dir, filename), mode='wb') as fout:
fout.write(file_bin)
return path.join(store_dir, filename)
def __init__(self, timeout=None, proxy=None, cacert=None, sessions=False):
if (timeout is not None) and not self.supports_feature('timeout'):
raise RuntimeError('timeout is not supported with urllib2 transport')
if proxy:
raise RuntimeError('proxy is not supported with urllib2 transport')
if cacert:
raise RuntimeError('cacert is not support with urllib2 transport')
handlers = []
if ((sys.version_info[0] == 2 and sys.version_info >= (2,7,9)) or
(sys.version_info[0] == 3 and sys.version_info >= (3,2,0))):
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
handlers.append(urllib2.HTTPSHandler(context=context))
if sessions:
handlers.append(urllib2.HTTPCookieProcessor(CookieJar()))
opener = urllib2.build_opener(*handlers)
self.request_opener = opener.open
self._timeout = timeout
def get_auth_token(apiKey, userid, password):
loginUrl = 'https://ivle.nus.edu.sg/api/login/?apikey=%s' % apiKey
data = urllib.request.urlopen(loginUrl).read()
if len(data) == 0:
raise InvalidAPIKeyException('API key is not valid.')
viewstate = re.search('__VIEWSTATE.+?value="(.+?)"', data)
if not viewstate:
# try setting viewstate to a hardcoded value if we fail trying to parse it
viewstate = '/wEPDwULLTEzODMyMDQxNjEPFgIeE1ZhbGlkYXRlUmVxdWVzdE1vZGUCARYCAgEPZBYEAgEPD2QWAh4Gb25ibHVyBQ91c2VySWRUb1VwcGVyKClkAgkPD2QWBB4Lb25tb3VzZW92ZXIFNWRvY3VtZW50LmdldEVsZW1lbnRCeUlkKCdsb2dpbmltZzEnKS5zcmM9b2ZmaW1nLnNyYzE7Hgpvbm1vdXNlb3V0BTRkb2N1bWVudC5nZXRFbGVtZW50QnlJZCgnbG9naW5pbWcxJykuc3JjPW9uaW1nLnNyYzE7ZBgBBR5fX0NvbnRyb2xzUmVxdWlyZVBvc3RCYWNrS2V5X18WAQUJbG9naW5pbWcxYTg4Q/LO3lNCB13iJpTeINmF1JQmGv61ni1TVgDIOII='
else:
viewstate = viewstate.group(1)
params = urllib.parse.urlencode({'__VIEWSTATE': viewstate, 'userid': userid, 'password': password})
cj = CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
userToken = opener.open(loginUrl, params).read()
if 'Login fail' in userToken or '</html>' in userToken:
raise InvalidLoginException('Login credentials are not valid.')
return userToken
# Adds authentication parameters to parameter list
def __init__(self, ssl_context, debug=False, proxies=None, no_proxy=None,
cookie=None, http_basicauth=None, headers=None):
"""
@param ssl_context: SSL context to use with this configuration
@type ssl_context: OpenSSL.SSL.Context
@param debug: if True, output debugging information
@type debug: bool
@param proxies: proxies to use for
@type proxies: dict with basestring keys and values
@param no_proxy: hosts for which a proxy should not be used
@type no_proxy: basestring
@param cookie: cookies to set for request
@type cookie: cookielib.CookieJar (python 3 - http.cookiejar)
@param http_basicauth: http authentication, or None
@type http_basicauth: tuple of (username,password)
@param headers: http headers
@type headers: dict
"""
self.ssl_context = ssl_context
self.debug = debug
self.proxies = proxies
self.no_proxy = no_proxy
self.cookie = cookie
self.http_basicauth = http_basicauth
self.headers = headers
def __init__(self, ssl_context, debug=False, proxies=None, no_proxy=None,
cookie=None, http_basicauth=None, headers=None):
"""
@param ssl_context: SSL context to use with this configuration
@type ssl_context: OpenSSL.SSL.Context
@param debug: if True, output debugging information
@type debug: bool
@param proxies: proxies to use for
@type proxies: dict with basestring keys and values
@param no_proxy: hosts for which a proxy should not be used
@type no_proxy: basestring
@param cookie: cookies to set for request
@type cookie: cookielib.CookieJar (python 3 - http.cookiejar)
@param http_basicauth: http authentication, or None
@type http_basicauth: tuple of (username,password)
@param headers: http headers
@type headers: dict
"""
self.ssl_context = ssl_context
self.debug = debug
self.proxies = proxies
self.no_proxy = no_proxy
self.cookie = cookie
self.http_basicauth = http_basicauth
self.headers = headers
def index(request):
if request.method == "GET":
try:
ssl._create_default_https_context = ssl._create_unverified_context
opener = wdf_urllib.build_opener(
wdf_urllib.HTTPCookieProcessor(CookieJar()))
wdf_urllib.install_opener(opener)
except:
pass
uuid = getUUID()
url = 'https://login.weixin.qq.com/qrcode/' + uuid
params = {
't': 'webwx',
'_': int(time.time()),
}
request = getRequest(url=url, data=urlencode(params))
response = wdf_urllib.urlopen(request)
context = {
'uuid': uuid,
'response': response.read(),
'delyou': '',
}
return render_to_response('index.html', context)
def __init__(self, ssl_context, debug=False, proxies=None, no_proxy=None,
cookie=None, http_basicauth=None, headers=None):
"""
@param ssl_context: SSL context to use with this configuration
@type ssl_context: OpenSSL.SSL.Context
@param debug: if True, output debugging information
@type debug: bool
@param proxies: proxies to use for
@type proxies: dict with basestring keys and values
@param no_proxy: hosts for which a proxy should not be used
@type no_proxy: basestring
@param cookie: cookies to set for request
@type cookie: cookielib.CookieJar (python 3 - http.cookiejar)
@param http_basicauth: http authentication, or None
@type http_basicauth: tuple of (username,password)
@param headers: http headers
@type headers: dict
"""
self.ssl_context = ssl_context
self.debug = debug
self.proxies = proxies
self.no_proxy = no_proxy
self.cookie = cookie
self.http_basicauth = http_basicauth
self.headers = headers
def login(username, password):
# Cookie Jar
cj = cookiejar.CookieJar()
s = requests.Session()
s.cookies = cj
br = mechanicalsoup.StatefulBrowser(soup_config={'features': 'lxml'}, session=s)
login_url = "http://cyclebabac.com/wp-login.php"
# Perform the actual login
br.open(login_url)
br.select_form('#loginform')
br['log'] = str(username)
br['pwd'] = str(password)
br.submit_selected()
return br
def __get_cookies(self, req):
cookies = cookiejar.CookieJar()
handler = request.HTTPCookieProcessor(cookies)
opener = request.build_opener(handler)
try:
with opener.open(req) as f:
if f.code == 200:
pattern = re.compile(r"<input.*?type='hidden'.*?name='csrfmiddlewaretoken'.*?value='(.*?)'.*>")
try:
self.csrfmiddlewaretoken = pattern.search(f.read().decode("utf-8")).group(1)
print("Achieved cookies and csrfmiddlewaretoken sucessfully")
except:
print("Achieved cookies sucessfully")
return cookies
else:
print("Lost cookies")
except error.URLError as e:
if hasattr(e, "reason"):
print ("We failed to reach a server. Please check your url and read the Reason")
print ("Reason: {}".format(e.reason))
elif hasattr(e, "code"):
print("The server couldn't fulfill the request.")
print("Error code: {}".format(e.code))
exit()
def load(self):
'''Load cookies into a cookiejar'''
cookie_jar = cookielib.CookieJar()
for cookie in self.get_cookies():
cookie_jar.set_cookie(cookie)
return cookie_jar
def load():
"""Try to load cookies from all supported browsers and return combined cookiejar
"""
cookie_jar = cookielib.CookieJar()
for cookie in sorted(_get_cookies(), key=lambda cookie: cookie.expires):
cookie_jar.set_cookie(cookie)
return cookie_jar
def __init__(self, username, password):
"""Performs login and save session cookie."""
# HTTPS Interface
# pylint: disable=C0103
cj = CookieJar()
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(cj),
urllib.request.HTTPSHandler())
self._loginV2(username, password)
self.id = self.getMe()['homes'][0]['id']
def __init__(self):
self.DEBUG = False
self.appid = 'wx782c26e4c19acffb'
self.uuid = ''
self.base_uri = ''
self.redirect_uri = ''
self.uin = ''
self.sid = ''
self.skey = ''
self.pass_ticket = ''
self.deviceId = 'e' + repr(random.random())[2:17]
self.BaseRequest = {}
self.synckey = ''
self.SyncKey = []
self.User = []
self.MemberList = []
self.ContactList = []
self.GroupList = []
self.autoReplyMode = False
self.syncHost = ''
self._handlers = dict((k, []) for k in self.message_types)
self._handlers['location'] = []
self._handlers['all'] = []
self._filters = dict()
opener = request.build_opener(request.HTTPCookieProcessor(CookieJar()))
opener.addheaders = [('User-agent', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/48.0.2564.109 Safari/537.36'),
('Referer','https://wx2.qq.com/')]
request.install_opener(opener)
def __init__(self, timeout=None, proxy=None, cacert=None, sessions=False):
if (timeout is not None) and not self.supports_feature('timeout'):
raise RuntimeError('timeout is not supported with urllib2 transport')
if proxy:
raise RuntimeError('proxy is not supported with urllib2 transport')
if cacert:
raise RuntimeError('cacert is not support with urllib2 transport')
self.request_opener = urllib2.urlopen
if sessions:
opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(CookieJar()))
self.request_opener = opener.open
self._timeout = timeout
def create_cookie_opener(self):
'''
????Cookie
:return: ????????opener
'''
cookie = cookiejar.CookieJar()
cookie_process = request.HTTPCookieProcessor(cookie)
opener = request.build_opener(cookie_process)
return opener
def __init__(self, timeout=None, proxy=None, cacert=None, sessions=False):
if (timeout is not None) and not self.supports_feature('timeout'):
raise RuntimeError('timeout is not supported with urllib2 transport')
if proxy:
raise RuntimeError('proxy is not supported with urllib2 transport')
if cacert:
raise RuntimeError('cacert is not support with urllib2 transport')
self.request_opener = urllib2.urlopen
if sessions:
opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(CookieJar()))
self.request_opener = opener.open
self._timeout = timeout
def test_cookie_redirect(self):
# cookies shouldn't leak into redirected requests
from http.cookiejar import CookieJar
from test.test_http_cookiejar import interact_netscape
cj = CookieJar()
interact_netscape(cj, "http://www.example.com/", "spam=eggs")
hh = MockHTTPHandler(302, "Location: http://www.cracker.com/\r\n\r\n")
hdeh = urllib.request.HTTPDefaultErrorHandler()
hrh = urllib.request.HTTPRedirectHandler()
cp = urllib.request.HTTPCookieProcessor(cj)
o = build_test_opener(hh, hdeh, hrh, cp)
o.open("http://www.example.com/")
self.assertFalse(hh.req.has_header("Cookie"))
def __init__(self, timeout=None, proxy=None, cacert=None, sessions=False):
if (timeout is not None) and not self.supports_feature('timeout'):
raise RuntimeError('timeout is not supported with urllib2 transport')
if proxy:
raise RuntimeError('proxy is not supported with urllib2 transport')
if cacert:
raise RuntimeError('cacert is not support with urllib2 transport')
self.request_opener = urllib2.urlopen
if sessions:
opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(CookieJar()))
self.request_opener = opener.open
self._timeout = timeout
def test_cookie_redirect(self):
# cookies shouldn't leak into redirected requests
from http.cookiejar import CookieJar
from test.test_http_cookiejar import interact_netscape
cj = CookieJar()
interact_netscape(cj, "http://www.example.com/", "spam=eggs")
hh = MockHTTPHandler(302, "Location: http://www.cracker.com/\r\n\r\n")
hdeh = urllib.request.HTTPDefaultErrorHandler()
hrh = urllib.request.HTTPRedirectHandler()
cp = urllib.request.HTTPCookieProcessor(cj)
o = build_test_opener(hh, hdeh, hrh, cp)
o.open("http://www.example.com/")
self.assertFalse(hh.req.has_header("Cookie"))
def to_pycookiejar(QtCookiejar):
cj = CookieJar()
for c in QtCookiejar.allCookies():
cj.set_cookie(to_py_cookie(c))
return cj
def __init__(self):
self.cookiejar = CookieJar()
self._cookie_processor = HTTPCookieProcessor(self.cookiejar)
self.form = None
self.url = "http://0.0.0.0:8080/"
self.path = "/"
self.status = None
self.data = None
self._response = None
self._forms = None
def baidu_pan_gen_cookies(url, post_data=None):
from http import cookiejar
cookiejar = cookiejar.CookieJar()
opener = request.build_opener(request.HTTPCookieProcessor(cookiejar))
resp = opener.open('http://pan.baidu.com')
if post_data != None:
resp = opener.open(url, bytes(parse.urlencode(post_data), 'utf-8'))
return cookjar2hdr(cookiejar)
def baidu_pan_gen_cookies(url, post_data=None):
from http import cookiejar
cookiejar = cookiejar.CookieJar()
opener = request.build_opener(request.HTTPCookieProcessor(cookiejar))
resp = opener.open('http://pan.baidu.com')
if post_data != None:
resp = opener.open(url, bytes(parse.urlencode(post_data), 'utf-8'))
return cookjar2hdr(cookiejar)
def test_cookie_redirect(self):
# cookies shouldn't leak into redirected requests
from http.cookiejar import CookieJar
from test.test_http_cookiejar import interact_netscape
cj = CookieJar()
interact_netscape(cj, "http://www.example.com/", "spam=eggs")
hh = MockHTTPHandler(302, "Location: http://www.cracker.com/\r\n\r\n")
hdeh = urllib.request.HTTPDefaultErrorHandler()
hrh = urllib.request.HTTPRedirectHandler()
cp = urllib.request.HTTPCookieProcessor(cj)
o = build_test_opener(hh, hdeh, hrh, cp)
o.open("http://www.example.com/")
self.assertFalse(hh.req.has_header("Cookie"))
def __init__(self, timeout=None, proxy=None, cacert=None, sessions=False):
if (timeout is not None) and not self.supports_feature('timeout'):
raise RuntimeError('timeout is not supported with urllib2 transport')
if proxy:
raise RuntimeError('proxy is not supported with urllib2 transport')
if cacert:
raise RuntimeError('cacert is not support with urllib2 transport')
self.request_opener = urllib2.urlopen
if sessions:
opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(CookieJar()))
self.request_opener = opener.open
self._timeout = timeout
def __init__(self, username="admin", password="admin", hostname="192.168.1.1"):
self.username = username
self.password = password
self.hostname = hostname
self.baseurl = "https://" + self.hostname
try:
self.gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
except AttributeError:
try:
self.gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1)
except AttributeError:
self.gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
self.cj = CookieJar()
self.opener = build_opener(HTTPCookieProcessor(self.cj), HTTPSHandler(context=self.gcontext))
self.loginresponse = self._login()
