我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用twisted.web.client.getPage()。
def validate(self, data, request): # We need only the body data = data[data.find('\r\n\r\n') + 4:] # Build the request for the validator service, using the # original request as the base headers = request.received_headers data, contentType = encodeFormData(self.arg, data) headers['content-type'] = contentType headers.pop('cookie', None) headers.pop('referer', None) headers.pop('host', None) return client.getPage( self.uri, method='POST', headers=headers, postdata=data )
def doRequest(self, service, body=missing, type=pyamf.AMF3, raw=False, decode=True): if not raw: if body is self.missing: body = [] else: body = [body] env = remoting.Envelope(type) request = remoting.Request(service, body=body) env['/1'] = request body = remoting.encode(env).getvalue() d = self.getPage(body) if decode: d.addCallback(lambda result: remoting.decode(result)) return d
def _next_request(self): while self.start_requests: try: request = next(self.start_requests) except StopIteration: self.start_requests = None else: self.scheduler.enqueue_request(request) while len(self.inprogress) < 5 and self.scheduler.size() > 0: # ??????5 request = self.scheduler.next_request() if not request: break self.inprogress.add(request) d = getPage(bytes(request.url, encoding='utf-8')) d.addBoth(self._handle_downloader_output, request) d.addBoth(lambda x, req: self.inprogress.remove(req), request) d.addBoth(lambda x: self._next_request()) if len(self.inprogress) == 0 and self.scheduler.size() == 0: self._closewait.callback(None)
def _get_page(self, method, url, data=None): try: response = yield getPage(url, method=method, headers={b'Content-Type': b'application/json'}, postdata=data) except Error as e: raise RestError(int(e.status), e.response) except Exception as e: raise RestError(-1, str(e)) try: obj = json.loads(response.decode('utf-8')) except Exception as e: raise RestError(-1, response) defer.returnValue(obj) # rpc wrapper, only usable for QRest class members
def test_afterFoundGet(self): """ Enabling unsafe redirection behaviour overwrites the method of redirected C{POST} requests with C{GET}. """ url = self.getURL('extendedRedirect?code=302') f = client.HTTPClientFactory(url, followRedirect=True, method=b"POST") self.assertFalse( f.afterFoundGet, "By default, afterFoundGet must be disabled") def gotPage(page): self.assertEqual( self.extendedRedirect.lastMethod, b"GET", "With afterFoundGet, the HTTP method must change to GET") d = client.getPage( url, followRedirect=True, afterFoundGet=True, method=b"POST") d.addCallback(gotPage) return d
def test_downloadAfterFoundGet(self): """ Passing C{True} for C{afterFoundGet} to L{client.downloadPage} invokes the same kind of redirect handling as passing that argument to L{client.getPage} invokes. """ url = self.getURL('extendedRedirect?code=302') def gotPage(page): self.assertEqual( self.extendedRedirect.lastMethod, b"GET", "With afterFoundGet, the HTTP method must change to GET") d = client.downloadPage(url, "downloadTemp", followRedirect=True, afterFoundGet=True, method=b"POST") d.addCallback(gotPage) return d
def test_getPageDeprecated(self): """ L{client.getPage} is deprecated. """ port = reactor.listenTCP( 0, server.Site(Data(b'', 'text/plain')), interface="127.0.0.1") portno = port.getHost().port self.addCleanup(port.stopListening) url = networkString("http://127.0.0.1:%d" % (portno,)) d = client.getPage(url) warningInfo = self.flushWarnings([self.test_getPageDeprecated]) self.assertEqual(len(warningInfo), 1) self.assertEqual(warningInfo[0]['category'], DeprecationWarning) self.assertEqual( warningInfo[0]['message'], "twisted.web.client.getPage was deprecated in " "Twisted 16.7.0; please use https://pypi.org/project/treq/ or twisted.web.client.Agent instead") return d.addErrback(lambda _: None)
def _get_page(self, url): def got_page(content): logging.debug("Cookies: %r" % self._cookies) return content def got_page_error(e, url): logging.error(url) log.err(e) return e url = url.encode("utf-8") if 'HLS_RESET_COOKIES' in os.environ.keys(): self._cookies = {} headers = {} if self.referer: headers['Referer'] = self.referer d = client.getPage(url, cookies=self._cookies, headers=headers) d.addCallback(got_page) d.addErrback(got_page_error, url) return d
def login(self): print("[GoogleReader] login") if not self.username or not self.password: return headers = {'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8'} data = { 'service': 'reader', 'Email': self.username, 'Passwd': self.password, 'source': 'enigma2-plugin-extensions-simplerss', 'continue': 'http://www.google.com/', } defer = Deferred() getPage('https://www.google.com/accounts/ClientLogin', method = 'POST', headers = headers, postdata = urlencode(data)).addCallback(self.loginFinished, defer).addErrback(self.loginFailed, defer) return defer
def _login(self, callback=None): self.debug("") if self._callScreen: self._callScreen.updateStatus(_("login")) if self._md5LoginTimestamp and ((time.time() - self._md5LoginTimestamp) < float(9.5 * 60)) and self._md5Sid != '0000000000000000': # new login after 9.5 minutes inactivity self.debug("[FritzCallFBF] renew timestamp: " + time.ctime(self._md5LoginTimestamp) + " time: " + time.ctime()) self._md5LoginTimestamp = time.time() callback(None) else: self.debug("not logged in or outdated login") # http://fritz.box/cgi-bin/webcm?getpage=../html/login_sid.xml parms = urlencode({'getpage': '../html/login_sid.xml'}) url = "http://%s/cgi-bin/webcm" % (config.plugins.FritzCall.hostname.value) self.debug("'" + url + "' parms: '" + parms + "'") getPage(url, method="POST", headers={'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata=parms).addCallback(lambda x: self._md5Login(callback, x)).addErrback(lambda x: self._oldLogin(callback, x))
def getCalls(self, callScreen, callback, callType): # # call sequence must be: # - login # - getPage -> _gotPageLogin # - loginCallback (_getCalls) # - getPage -> _getCalls1 self.debug("") self._callScreen = callScreen self._callType = callType if (time.time() - self._callTimestamp) > 180: self.debug("outdated data, login and get new ones: " + time.ctime(self._callTimestamp) + " time: " + time.ctime()) self._callTimestamp = time.time() self._login(lambda x: self._getCalls(callback, x)) elif not self._callList: self.debug("time is ok, but no callList") self._getCalls1(callback) else: self.debug("time is ok, callList is ok") self._gotPageCalls(callback)
def _selectFritzBoxPhonebook(self, html): # first check for login error if html: start = html.find('<p class="errorMessage">FEHLER: ') if start != -1: start = start + len('<p class="errorMessage">FEHLER: ') self._errorLoad('Login: ' + html[start, html.find('</p>', start)]) return # look for phonebook called dreambox or Dreambox parms = urlencode({ 'sid':self._md5Sid, }) url = "http://%s/fon_num/fonbook_select.lua" % (config.plugins.FritzCall.hostname.value) self.debug("[FritzCallFBF_05_27] _selectPhonebook: '" + url + "' parms: '" + parms + "'") getPage(url, method = "POST", agent = USERAGENT, headers = {'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(self._loadFritzBoxPhonebook).addErrback(self._errorLoad)
def _loadFritzBoxPhonebook(self, html): # Firmware 05.27 onwards # look for phonebook called [dD]reambox and get bookid found = re.match(r'.*<label for="uiBookid:([\d]+)">' + config.plugins.FritzCall.fritzphonebookName.value, html, re.S) if found: bookid = found.group(1) self.debug("[FritzCallFBF_05_27] _loadFritzBoxPhonebook: found dreambox phonebook %s", bookid) else: bookid = 1 # http://192.168.178.1/fon_num/fonbook_list.lua?sid=2faec13b0000f3a2 parms = urlencode({ 'bookid':bookid, 'sid':self._md5Sid, }) url = "http://%s/fon_num/fonbook_list.lua" % (config.plugins.FritzCall.hostname.value) self.debug("[FritzCallFBF_05_27] _loadFritzBoxPhonebookNew: '" + url + "' parms: '" + parms + "'") getPage(url, method = "POST", agent = USERAGENT, headers = {'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(self._parseFritzBoxPhonebook).addErrback(self._errorLoad)
def _getInfo(self, callback, html): self.debug("[FritzCallFBF_05_27] _getInfo: verify login") if html: start = html.find('<p class="errorMessage">FEHLER: ') if start != -1: start = start + len('<p class="errorMessage">FEHLER: ') self._errorGetInfo('Login: ' + html[start, html.find('</p>', start)]) return self._readBlacklist() url = "http://%s/home/home.lua" % config.plugins.FritzCall.hostname.value parms = urlencode({ 'sid':self._md5Sid }) self.debug("[FritzCallFBF_05_27] _getInfo url: '" + url + "' parms: '" + parms + "'") getPage(url, method = "POST", agent = USERAGENT, headers = { 'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(lambda x:self._okGetInfo(callback, x)).addErrback(self._errorGetInfo)
def _loadFritzBoxPhonebook(self, html, md5Sid): # Firmware 05.27 onwards # look for phonebook called [dD]reambox and get bookid found = re.match(r'.*<label for="uiBookid:([\d]+)">' + config.plugins.FritzCall.fritzphonebookName.value, html, re.S) if found: bookid = found.group(1) else: bookid = 1 self.debug("phonebook %s", bookid) # http://192.168.178.1/fon_num/fonbook_list.lua?sid=2faec13b0000f3a2 parms = urlencode({ 'bookid':bookid, 'sid':md5Sid, 'cancel':'', 'apply':'uiApply', }) url = "http://%s/fon_num/fonbook_select.lua" % (config.plugins.FritzCall.hostname.value) self.debug(url + "?" + parms) getPage(url, method = "POST", agent = USERAGENT, headers = {'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(self._parseFritzBoxPhonebook, md5Sid).addErrback(self._errorLoad, md5Sid)
def _dial(self, number, md5Sid): url = "http://%s/cgi-bin/webcm" % config.plugins.FritzCall.hostname.value parms = urlencode({ 'getpage':'../html/de/menus/menu2.html', 'var:pagename':'fonbuch', 'var:menu':'home', 'telcfg:settings/UseClickToDial':'1', 'telcfg:settings/DialPort':config.plugins.FritzCall.extension.value, 'telcfg:command/Dial':number, 'sid':md5Sid }) self.info("url: " + url + "?" + parms) getPage(url, method = "POST", agent = USERAGENT, headers = { 'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(self._okDial, md5Sid).addErrback(self._errorDial, md5Sid)
def _getInfo(self, callback, md5Sid): self.debug("verify login") self._login(self._readBlacklist) url = "http://%s/home/home.lua" % config.plugins.FritzCall.hostname.value parms = urlencode({ 'sid':md5Sid }) self.debug("url: " + url + "?" + parms) getPage(url, method = "POST", agent = USERAGENT, headers = { 'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(lambda x:self._okGetInfo(callback, x, md5Sid)).addErrback(self._errorGetInfo, md5Sid)
def _reset(self, md5Sid): if self._callScreen: self._callScreen.close() url = "http://%s/system/reboot.lua" % config.plugins.FritzCall.hostname.value parms = urlencode({ 'reboot':'', 'sid':md5Sid }) self.debug("url: " + url + "?" + parms) getPage(url, method = "POST", agent = USERAGENT, headers = { 'Content-Type': "application/x-www-form-urlencoded"}, postdata = parms).addCallback(self._okReset, md5Sid).addErrback(self._errorReset, md5Sid)
def _getInfo(self, callback, md5Sid): self.debug("verify login") self._login(self._readBlacklist) url = "http://%s/data.lua" % config.plugins.FritzCall.hostname.value parms = urlencode({ 'sid':md5Sid, 'page':'overview', 'type':'all' }) self.debug("url: " + url + "?" + parms) getPage(url, method = "POST", agent = USERAGENT, headers = { 'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(lambda x:self._okGetInfo(callback, x, md5Sid)).addErrback(self._errorGetInfo, md5Sid)
def _readBlacklist(self, md5Sid): # http://fritz.box/cgi-bin/webcm?getpage=../html/de/menus/menu2.html&var:lang=de&var:menu=fon&var:pagename=sperre # https://217.245.196.140:699/data.lua?xhr=1&sid=e8fcf4f9a9186070&lang=de&no_sidrenew=&page=callLock url = "http://%s/data.lua" % config.plugins.FritzCall.hostname.value parms = urlencode({ 'sid':md5Sid, 'page':'callLock' }) self.debug("url: " + url + "?" + parms) getPage(url, method = "POST", agent = USERAGENT, headers = { 'Content-Type': "application/x-www-form-urlencoded", 'Content-Length': str(len(parms))}, postdata = parms).addCallback(self._okBlacklist, md5Sid).addErrback(self._errorBlacklist, md5Sid)
def publicGetPage(*args, **kwargs): d = getPage(*args, **kwargs) return PublicDeferred(d)
def getPage(self, data=None, **kwargs): kwargs.setdefault('method', 'POST') kwargs['postdata'] = data return client.getPage("http://127.0.0.1:%d/" % (self.port,), **kwargs)
def test_invalid_method(self): """ A classic GET on the xml server should return a NOT_ALLOWED. """ d = self.getPage(method='GET') d = self.assertFailure(d, error.Error) d.addCallback( lambda exc: self.assertEqual(int(exc.args[0]), http.NOT_ALLOWED)) return d
def test_bad_content(self): d = self.getPage('spamandeggs') d = self.assertFailure(d, error.Error) d.addCallback( lambda exc: self.assertEqual(int(exc.args[0]), http.BAD_REQUEST)) return d
def callRemote(self, method, *args, **kwargs): payload = SOAPpy.buildSOAP(args=args, kw=kwargs, method=method, header=self.header, namespace=self.namespace) return client.getPage(self.url, postdata=payload, method="POST", headers={'content-type': 'text/xml', 'SOAPAction': method} ).addCallback(self._cbGotResult)
def testCGI(self): cgiFilename = os.path.abspath(self.mktemp()) cgiFile = file(cgiFilename, 'wt') cgiFile.write(DUMMY_CGI) cgiFile.close() portnum = self.startServer(cgiFilename) d = client.getPage("http://localhost:%d/cgi" % portnum) d.addCallback(self._testCGI_1) return d
def testReadEmptyInput(self): cgiFilename = os.path.abspath(self.mktemp()) cgiFile = file(cgiFilename, 'wt') cgiFile.write(READINPUT_CGI) cgiFile.close() portnum = self.startServer(cgiFilename) d = client.getPage("http://localhost:%d/cgi" % portnum) d.addCallback(self._testReadEmptyInput_1) return d
def testReadInput(self): cgiFilename = os.path.abspath(self.mktemp()) cgiFile = file(cgiFilename, 'wt') cgiFile.write(READINPUT_CGI) cgiFile.close() portnum = self.startServer(cgiFilename) d = client.getPage("http://localhost:%d/cgi" % portnum, method="POST", postdata="Here is your stdin") d.addCallback(self._testReadInput_1) return d
def testReadAllInput(self): cgiFilename = os.path.abspath(self.mktemp()) cgiFile = file(cgiFilename, 'wt') cgiFile.write(READALLINPUT_CGI) cgiFile.close() portnum = self.startServer(cgiFilename) d = client.getPage("http://localhost:%d/cgi" % portnum, method="POST", postdata="Here is your stdin") d.addCallback(self._testReadAllInput_1) return d
def testPayload(self): s = "0123456789" * 10 return client.getPage(self.getURL("payload"), postdata=s ).addCallback(self.assertEquals, s )
def testBrokenDownload(self): # test what happens when download gets disconnected in the middle d = client.getPage(self.getURL("broken")) d = self.assertFailure(d, client.PartialDownloadError) d.addCallback(lambda exc: self.assertEquals(exc.response, "abc")) return d
def testHostHeader(self): # if we pass Host header explicitly, it should be used, otherwise # it should extract from url return defer.gatherResults([ client.getPage(self.getURL("host")).addCallback(self.assertEquals, "127.0.0.1"), client.getPage(self.getURL("host"), headers={"Host": "www.example.com"}).addCallback(self.assertEquals, "www.example.com")])
def testGetPage(self): return client.getPage(self.getURL("file")).addCallback(self.assertEquals, "0123456789")
def testTimeoutNotTriggering(self): # Test that when the timeout doesn't trigger, things work as expected. d = client.getPage(self.getURL("wait"), timeout=100) d.addCallback(self.assertEquals, "hello!!!") return d
def testServerError(self): return self._downloadTest(lambda url: client.getPage(self.getURL(url)))
def testRedirect(self): return client.getPage(self.getURL("redirect")).addCallback(self._cbRedirect)
def _cbRedirect(self, pageData): self.assertEquals(pageData, "0123456789") d = self.assertFailure( client.getPage(self.getURL("redirect"), followRedirect=0), error.PageRedirect) d.addCallback(self._cbCheckLocation) return d
def testHoppingAround(self): return client.getPage(self.getHTTP("one") ).addCallback(self.assertEquals, "FOUND IT!" )
def testNoCookies(self): return client.getPage(self.getHTTP("cookiemirror") ).addCallback(self.assertEquals, "[]" )
def testRawNoCookies(self): return client.getPage(self.getHTTP("rawcookiemirror") ).addCallback(self.assertEquals, "None" )
def testRawSomeCookies(self): cookies = {'foo': 'bar', 'baz': 'quux'} return client.getPage(self.getHTTP("rawcookiemirror"), cookies=cookies ).addCallback(self.assertEquals, "'foo=bar; baz=quux'" )
def _getPage(self): d = client.getPage(self.url) d.addErrback(self.noPage) d.addCallback(self.page) self.call = None
def getList(self, callback, errback): self.urlbase = "%s/%s/%s/images/" % (self.feed_base, self.OE_vers, self.box) print "[getList]", self.urlbase self.callback = callback self.errback = errback client.getPage(self.urlbase).addCallback(self.feed_finished).addErrback(self.feed_failed)
def getMD5(self): url = "http://www.dreamboxupdate.com/download/opendreambox/dreambox-nfiflasher-%s-md5sums" % self.box client.getPage(url).addCallback(self.md5sums_finished).addErrback(self.feed_failed)
def nfo_download(self, branch, idx): nfourl = (self.feedlists[branch][idx][1])[:-4]+".nfo" self.nfofilename = (self.feedlists[branch][idx][0])[:-4]+".nfo" print "[check_for_NFO]", nfourl client.getPage(nfourl).addCallback(self.nfo_finished).addErrback(self.nfo_failed)
def download(self, url): return getPage(url, timeout=20, headers={'Accept': 'application/json'})
def call(self, method, **kwargs): """ Call the given remote method with the given arguments, as keywords. Returns a deferred, called with SOAPpy structure representing the soap response. @param method: The method name to call, eg. 'GetExternalIP' @param kwargs: The parameters of the call, as keywords @return: A deferred called with the external ip address of this host @rtype: L{twisted.internet.defer.Deferred} """ payload = SOAPpy.buildSOAP(method=method, config=Config, namespace=self._prefix, kw=kwargs) # Here begins the nasty hack payload = payload.replace( # Upnp wants s: instead of SOAP-ENV 'SOAP-ENV','s').replace( # Doesn't seem to like these encoding stuff 'xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/"', '').replace( 'SOAP-ENC:root="1"', '').replace( # And it wants u: instead of ns1 namespace for arguments.. 'ns1','u') logging.debug("SOAP Payload:\n%s", payload) return client.getPage(self._url, postdata=payload, method="POST", headers={'content-type': 'text/xml', 'SOAPACTION': '%s#%s' % (self._prefix, method)} ).addCallbacks(self._got_page, self._got_error)
def _http_do(url, headers, timeout, method, params): id_ = 0 try: data = yield client.getPage( url=url, method='POST', headers=dict(headers, **{'Content-Type': 'application/json'}), postdata=json.dumps({ 'jsonrpc': '2.0', 'method': method, 'params': params, 'id': id_, }), timeout=timeout, ) except error.Error, e: try: resp = json.loads(e.response) except: raise e else: resp = json.loads(data) if resp['id'] != id_: raise ValueError('invalid id') if 'error' in resp and resp['error'] is not None: raise Error_for_code(resp['error']['code'])(resp['error']['message'], resp['error'].get('data', None)) defer.returnValue(resp['result'])
def async_url_read(url, timeout=0, **kwargs): if url.startswith('http'): content = getPage(encode(url), timeout=timeout) else: content = async_read_file(url, StringTransport(), **kwargs) return content
def makeAPICall(self, path, method="POST", params=None): if params: postdata = json.dumps(params) else: postdata = None uri = networkString("http://127.0.0.1:%d/API/%s" % ( self.plainPortno, path)) return client.getPage( uri, method=method, timeout=1, postdata=postdata, headers={'X-Bitmask-Auth': 'aaa'})
