# -*- coding: utf-8 -*- """ Created on 2018-07-26 11:40:28 --------- @summary: --------- @author: Boris @email: [email protected] """ import datetime import os import re import time from urllib.parse import urlparse, urlunparse, urljoin from bs4 import UnicodeDammit, BeautifulSoup from requests.cookies import RequestsCookieJar from requests.models import Response as res from w3lib.encoding import http_content_type_encoding, html_body_declared_encoding from feapder.network.selector import Selector from feapder.utils.log import log FAIL_ENCODING = "ISO-8859-1" # html 源码中的特殊字符，需要删掉，否则会影响etree的构建 SPECIAL_CHARACTERS = [ # 移除控制字符 全部字符列表 https://zh.wikipedia.org/wiki/%E6%8E%A7%E5%88%B6%E5%AD%97%E7%AC%A6 "[\x00-\x08\x0B\x0C\x0E-\x1F\x7F-\x9F]" ] SPECIAL_CHARACTER_PATTERNS = [ re.compile(special_character) for special_character in SPECIAL_CHARACTERS ] class Response(res): def __init__(self, response): super(Response, self).__init__() self.__dict__.update(response.__dict__) self._cached_selector = None self._cached_text = None self._cached_json = None self._encoding = None self.encoding_errors = "strict" # strict / replace / ignore @classmethod def from_dict(cls, response_dict): """ 利用字典获取Response对象 @param response_dict: 原生的response.__dict__ @return: """ cookie_jar = RequestsCookieJar() cookie_jar.update(other=response_dict["cookies"]) response_dict["cookies"] = cookie_jar response_dict["elapsed"] = datetime.timedelta( 0, 0, response_dict["elapsed"] ) # 耗时 response_dict["connection"] = None response_dict["_content_consumed"] = True response = res() response.__dict__.update(response_dict) return cls(response) @property def to_dict(self): response_dict = { "_content": self.content, "cookies": self.cookies.get_dict(), "encoding": self.encoding, "headers": self.headers, "status_code": self.status_code, "elapsed": self.elapsed.microseconds, # 耗时 "url": self.url, } return response_dict def __clear_cache(self): self.__dict__["_cached_selector"] = None self.__dict__["_cached_text"] = None self.__dict__["_cached_json"] = None @property def encoding(self): """ 编码优先级：自定义编码 > header中编码 > 页面编码 > 根据content猜测的编码 """ self._encoding = ( self._encoding or self._headers_encoding() or self._body_declared_encoding() or self.apparent_encoding ) return self._encoding @encoding.setter def encoding(self, val): self.__clear_cache() self._encoding = val code = encoding def _headers_encoding(self): """ 从headers获取头部charset编码 """ content_type = self.headers.get("Content-Type") or self.headers.get( "content-type" ) if content_type: return ( http_content_type_encoding(content_type) or "utf-8" if "application/json" in content_type else None ) def _body_declared_encoding(self): """ 从html xml等获取 """ return html_body_declared_encoding(self.content) def _get_unicode_html(self, html): if not html or not isinstance(html, bytes): return html converted = UnicodeDammit(html, is_html=True) if not converted.unicode_markup: raise Exception( "Failed to detect encoding of article HTML, tried: %s" % ", ".join(converted.tried_encodings) ) html = converted.unicode_markup return html def _make_absolute(self, link): """Makes a given link absolute.""" try: link = link.strip() # Parse the link with stdlib. parsed = urlparse(link)._asdict() # If link is relative, then join it with base_url. if not parsed["netloc"]: return urljoin(self.url, link) # Link is absolute; if it lacks a scheme, add one from base_url. if not parsed["scheme"]: parsed["scheme"] = urlparse(self.url).scheme # Reconstruct the URL to incorporate the new scheme. parsed = (v for v in parsed.values()) return urlunparse(parsed) except Exception as e: log.error( "Invalid URL <{}> can't make absolute_link. exception: {}".format( link, e ) ) # Link is absolute and complete with scheme; nothing to be done here. return link def _absolute_links(self, text): regexs = [ r'(<(?i)a.*?href\s*?=\s*?["\'])(.+?)(["\'])', # a r'(<(?i)img.*?src\s*?=\s*?["\'])(.+?)(["\'])', # img r'(<(?i)link.*?href\s*?=\s*?["\'])(.+?)(["\'])', # css r'(<(?i)script.*?src\s*?=\s*?["\'])(.+?)(["\'])', # js ] for regex in regexs: def replace_href(text): # html = text.group(0) link = text.group(2) absolute_link = self._make_absolute(link) # return re.sub(regex, r'\1{}\3'.format(absolute_link), html) # 使用正则替换，个别字符不支持。如该网址源代码http://permit.mep.gov.cn/permitExt/syssb/xxgk/xxgk!showImage.action?dataid=0b092f8115ff45c5a50947cdea537726 return text.group(1) + absolute_link + text.group(3) text = re.sub(regex, replace_href, text, flags=re.S) return text def _del_special_character(self, text): """ 删除特殊字符 """ for special_character_pattern in SPECIAL_CHARACTER_PATTERNS: text = special_character_pattern.sub("", text) return text @property def __text(self): """Content of the response, in unicode. If Response.encoding is None, encoding will be guessed using ``chardet``. The encoding of the response content is determined based solely on HTTP headers, following RFC 2616 to the letter. If you can take advantage of non-HTTP knowledge to make a better guess at the encoding, you should set ``r.encoding`` appropriately before accessing this property. """ if not self.content: return "" # Decode unicode from given encoding. try: content = str(self.content, self.encoding, errors=self.encoding_errors) except (LookupError, TypeError): # A LookupError is raised if the encoding was not found which could # indicate a misspelling or similar mistake. # # A TypeError can be raised if encoding is None # # So we try blindly encoding. content = str(self.content, errors=self.encoding_errors) return content @property def text(self): if self._cached_text is None: if self.encoding and self.encoding.upper() != FAIL_ENCODING: try: self._cached_text = self.__text except UnicodeDecodeError: self._cached_text = self._get_unicode_html(self.content) else: self._cached_text = self._get_unicode_html(self.content) if self._cached_text: self._cached_text = self._absolute_links(self._cached_text) self._cached_text = self._del_special_character(self._cached_text) return self._cached_text @text.setter def text(self, html): self._cached_text = html self._cached_text = self._absolute_links(self._cached_text) self._cached_text = self._del_special_character(self._cached_text) self._cached_selector = Selector(self.text) @property def json(self, **kwargs): if self._cached_json is None: self.encoding = self.encoding or "utf-8" self._cached_json = super(Response, self).json(**kwargs) return self._cached_json @property def content(self): content = super(Response, self).content return content @property def is_html(self): content_type = self.headers.get("Content-Type", "") if "text/html" in content_type: return True else: return False @property def selector(self): if self._cached_selector is None: self._cached_selector = Selector(self.text) return self._cached_selector def bs4(self, features="html.parser"): soup = BeautifulSoup(self.text, features) return soup def extract(self): return self.selector.get() def xpath(self, query, **kwargs): return self.selector.xpath(query, **kwargs) def css(self, query): return self.selector.css(query) def re(self, regex, replace_entities=False): """ @summary: 正则匹配 注意：网页源码