处理100~199段状态码,针对不同的服务响应进行操作.

        Parameters:
            (response): - 响应的python字典形式数据

        Return:
            (bool): - 准确地说没有错误就会返回True
323073
def _server_response_handler(self, response: Dict[str, Any]):
        
        code = response.get("CODE")
        if code == 100:
            if self.debug:
                print("auth succeed")
            self._login_fut.set_result(response)
        if code == 101:
            if self.debug:
                print('pong')
        return True
850368
def QA_util_code_tostr(code):
    
    if isinstance(code, int):
        return "{:>06d}".format(code)
    if isinstance(code, str):
        # 聚宽股票代码格式 '600000.XSHG'
        # 掘金股票代码格式 'SHSE.600000'
        # Wind股票代码格式 '600000.SH'
        # 天软股票代码格式 'SH600000'
        if len(code) == 6:
            return code
        if len(code) == 8:
            # 天软数据
            return code[-6:]
        if len(code) == 9:
            return code[:6]
        if len(code) == 11:
            if code[0] in ["S"]:
                return code.split(".")[1]
            return code.split(".")[0]
        raise ValueError("错误的股票代码格式")
    if isinstance(code, list):
        return QA_util_code_to_str(code[0])
413650
def run_check200(_):
    '''
    
    '''

    tstr = ''
    idx = 1
    for kind in config.router_post.keys():
        posts = MPost.query_all(kind=kind, limit=20000)
        for post in posts:
            the_url0 = '{site_url}/{kind_url}/{uid}'.format(
                site_url=config.SITE_CFG['site_url'],
                kind_url=config.router_post[post.kind],
                uid=post.uid)

            the_url = '{site_url}/{kind_url}/_edit/{uid}'.format(
                site_url=config.SITE_CFG['site_url'],
                kind_url=config.router_post[post.kind],
                uid=post.uid)
            req = requests.get(the_url0)

            if req.status_code == 200:
                pass
            else:
                print(the_url0)
                tstr = tstr + DT_STR.format(idx=str(idx).zfill(2), url0=the_url0, code=req.status_code, edit_link=the_url)
                idx = idx + 1

    time_local = time.localtime(timestamp())
    with open('xx200_{d}.html'.format(d=str(time.strftime("%Y_%m_%d", time_local))), 'w') as fileo:
        fileo.write(HTML_TMPL.format(cnt=tstr))
    print('Checking 200 finished.')
417075
def _decode_response(self, ip_address):
        
        # Reverse the IP, reassign the octets to integers
        vt, ts, days, rc = [int(o) for o in ip_address.split('.')[::-1]]

        # 127 reflects a valid query response, all others are errors
        if rc != 127:
            raise ValueError('Invalid Response Code: {}'.format(rc))

        # Build a list of visitor types since one IP can be multiple
        visitor_types = []
        if vt & COMMENT_SPAMMER:
            visitor_types.append(COMMENT_SPAMMER)
        if vt & HARVESTER:
            visitor_types.append(HARVESTER)
        if vt & SUSPICIOUS:
            visitor_types.append(SUSPICIOUS)

        # Return the response dictionary
        return {'days_since_last_activity': days if vt else None,
                'name': None if vt else SEARCH_ENGINES[ts],
                'threat_score': ts if vt else None,
                'type': visitor_types if vt else [SEARCH_ENGINE]}
1318544
def fetch(self, url, timeout=None):
        

        # ISO-8859-1 is the default encoding for text files per the specs for
        # HTTP 1.0 (RFC 1945 sec 3.6.1) and HTTP 1.1 (RFC 2616 sec 3.7.1).
        # ref: http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.7.1
        encoding = "iso-8859-1"
        content = ""
        expires_header = None
        content_type_header = None
        self._response_code = 0
        self._source_url = url

        if self.user_agent:
            req = urllib_request.Request(url, None, 
                                         { 'User-Agent' : self.user_agent })
        else:
            req = urllib_request.Request(url)

        try:
            if timeout:
                f = urllib_request.urlopen(req, timeout=timeout)
            else:
                f = urllib_request.urlopen(req)

            content = f.read(MAX_FILESIZE)
            # As of Python 2.5, f.info() looks like it returns the HTTPMessage
            # object created during the connection. 
            expires_header = f.info().get("expires")
            content_type_header = f.info().get("Content-Type")
            # As of Python 2.4, this file-like object reports the response 
            # code, too. 
            if hasattr(f, "code"):
                self._response_code = f.code
            else:
                self._response_code = 200
            f.close()
        except urllib_error.URLError:
            # This is a slightly convoluted way to get the error instance,
            # but it works under Python 2 & 3. 
            error_instance = sys.exc_info()
            if len(error_instance) > 1:
                error_instance = error_instance[1]
            if hasattr(error_instance, "code"):
                self._response_code = error_instance.code
                
        # MK1996 section 3.4 says, "...robots should take note of Expires 
        # header set by the origin server. If no cache-control directives 
        # are present robots should default to an expiry of 7 days".
        
        # This code is lazy and looks at the Expires header but not 
        # Cache-Control directives.
        self.expiration_date = None
        if self._response_code >= 200 and self._response_code < 300:
            # All's well.
            if expires_header:
                self.expiration_date = email_utils.parsedate_tz(expires_header)
                
                if self.expiration_date:
                    # About time zones -- the call to parsedate_tz() returns a
                    # 10-tuple with the time zone offset in the 10th element. 
                    # There are 3 valid formats for HTTP dates, and one of 
                    # them doesn't contain time zone information. (UTC is 
                    # implied since all HTTP header dates are UTC.) When given
                    # a date that lacks time zone information, parsedate_tz() 
                    # returns None in the 10th element. mktime_tz() interprets
                    # None in the 10th (time zone) element to mean that the 
                    # date is *local* time, not UTC. 
                    # Therefore, if the HTTP timestamp lacks time zone info 
                    # and I run that timestamp through parsedate_tz() and pass
                    # it directly to mktime_tz(), I'll get back a local 
                    # timestamp which isn't what I want. To fix this, I simply
                    # convert a time zone of None to zero. It's much more 
                    # difficult to explain than to fix. =)
                    # ref: http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.3.1
                    if self.expiration_date[9] == None: 
                        self.expiration_date = self.expiration_date[:9] + (0,)
                
                    self.expiration_date = email_utils.mktime_tz(self.expiration_date)
                    if self.use_local_time: 
                        # I have to do a little more converting to get this 
                        # UTC timestamp into localtime.
                        self.expiration_date = time.mktime(time.gmtime(self.expiration_date)) 
                #else:
                    # The expires header was garbage.

        if not self.expiration_date: self.expiration_date = self._now() + SEVEN_DAYS

        if (self._response_code >= 200) and (self._response_code < 300):
            # All's well.
            media_type, encoding = _parse_content_type_header(content_type_header)
            # RFC 2616 sec 3.7.1 -- 
            # When no explicit charset parameter is provided by the sender, 
            # media subtypes  of the "text" type are defined to have a default
            # charset value of "ISO-8859-1" when received via HTTP.
            # http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.7.1
            if not encoding: 
                encoding = "iso-8859-1"
        elif self._response_code in (401, 403):
            # 401 or 403 ==> Go away or I will taunt you a second time! 
            # (according to MK1996)
            content = "User-agent: *\nDisallow: /\n"
        elif self._response_code == 404:
            # No robots.txt ==> everyone's welcome
            content = ""
        else:        
            # Uh-oh. I punt this up to the caller. 
            _raise_error(urllib_error.URLError, self._response_code)

        if ((PY_MAJOR_VERSION == 2) and isinstance(content, str)) or \
           ((PY_MAJOR_VERSION > 2)  and (not isinstance(content, str))):
            # This ain't Unicode yet! It needs to be.
            
            # Unicode decoding errors are another point of failure that I punt 
            # up to the caller.
            try:
                content = content.decode(encoding)
            except UnicodeError:
                _raise_error(UnicodeError,
                "Robots.txt contents are not in the encoding expected (%s)." % encoding)
            except (LookupError, ValueError):
                # LookupError ==> Python doesn't have a decoder for that encoding.
                # One can also get a ValueError here if the encoding starts with 
                # a dot (ASCII 0x2e). See Python bug 1446043 for details. This 
                # bug was supposedly fixed in Python 2.5.
                _raise_error(UnicodeError,
                        "I don't understand the encoding \"%s\"." % encoding)
        
        # Now that I've fetched the content and turned it into Unicode, I 
        # can parse it.
        self.parse(content)
164151
def parse229(resp) # :nodoc:
      if resp[0, 3] != "229"
        raise FTPReplyError, resp
      end
      if m = /\((?<d>[!-~])\k<d>\k<d>(?<port>\d+)\k<d>\)/.match(resp)
        return @sock.peeraddr[3], m["port"].to_i
      else
        raise FTPProtoError, resp
      end
    end
566108
def text(
    body, status=200, headers=None, content_type="text/plain; charset=utf-8"
):
    
    return HTTPResponse(
        body, status=status, headers=headers, content_type=content_type
    )
62309
def handle_response(response)
      case response.code
      when 200..299
        response
      else
        if response.request.format == :json
          raise FreebaseAPI::ServiceError.new(response['error'])
        else
          raise FreebaseAPI::NetError.new('code' => response.code, 'message' => response.response.message)
        end
      end
    end
836253
def get_status_from_location(self, response):
        
        self._raise_if_bad_http_status_and_method(response)
        code = response.status_code
        if code == 202:
            self.status = "InProgress"
        else:
            self.status = 'Succeeded'
            if self._is_empty(response):
                self.resource = None
            else:
                self.resource = self._deserialize(response)
1262070
def teredo(self):
        
        if (self._ip >> 96) != 0x20010000:
            return None
        return (IPv4Address((self._ip >> 64) & 0xFFFFFFFF),
                IPv4Address(~self._ip & 0xFFFFFFFF))
262180
def get_question_results(self, number: str, username: str) -> dict:
        
        try:
            # 操作所需資訊
            params = {
                'questionID': number,
                'studentID': username
            }
            # 取得資料
            response = self.__session.get(
                self.__url + '/CheckResult.jsp', params=params, timeout=0.5, verify=False)
            soup = BeautifulSoup(response.text, 'html.parser')
            # 整理指定用戶之測試結果
            results = {}
            for tag in soup.find_all('tr'):
                # 取得測試結果
                result = tag.find_all('td')
                # 跳過標題列
                if result[0].get_text().strip() != '測試編號':
                    index = result[0].get_text().strip()
                    status = result[1].get_text().strip()
                    results[index] = status
            # 回傳結果
            return results

        except requests.exceptions.Timeout:
            return {'Timeout': 'Timeout'}
863640
def valid_hotp(self, code, last=0, trials=100):
        
        if not valid_code(code):
            return False

        code = bytes(int(code))
        for i in range(last + 1, last + trials + 1):
            if compare_digest(bytes(self.hotp(counter=i)), code):
                return i
        return False
1072785
def get(self, *args, **kwargs):
        

        self._parameters.query = kwargs.pop('query', {}) if len(args) == 0 else args[0]
        self._parameters.limit = kwargs.pop('limit', 10000)
        self._parameters.offset = kwargs.pop('offset', 0)
        self._parameters.fields = kwargs.pop('fields', kwargs.pop('fields', []))

        return self._get_response('GET', stream=kwargs.pop('stream', False))
469623
def _raise_if_bad_http_status_and_method(self, response):
        
        code = response.status_code
        if code in {200, 202} or \
           (code == 201 and self.method in {'PUT', 'PATCH'}) or \
           (code == 204 and self.method in {'DELETE', 'POST'}):
            return
        raise BadStatus(
            "Invalid return status for {!r} operation".format(self.method))
743433
protected function isResponseOk()
    {
        $status = $this->rawResponse->status;
        if ((int)(intval($status) / 100) == 2 || (int)(intval($status)) === 404) {
            return true;
        }
        return false;
    }
88817
private boolean isRateLimitReached(ResponseEntity<String> response) {
        HttpHeaders header = response.getHeaders();
        List<String> limit = header.get("X-RateLimit-Remaining");
        boolean rateLimitReached =  CollectionUtils.isEmpty(limit) ? false : Integer.valueOf(limit.get(0)) < settings.getRateLimitThreshold();
        if (rateLimitReached) {
            LOG.error("Github rate limit reached. Threshold =" + settings.getRateLimitThreshold() + ". Current remaining ="+Integer.valueOf(limit.get(0)));
        }
        return rateLimitReached;
    }
1211605
def get_response(self):
        
        while not self.comm_chan.recv_ready():
            time.sleep(0.5)
        return self.comm_chan.recv(2048)