forked from psf/requests-html
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathrequests_html.py
322 lines (240 loc) · 9.36 KB
/
requests_html.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
import asyncio
from urllib.parse import urlparse, urlunparse
import pyppeteer
import requests
from pyquery import PyQuery
from fake_useragent import UserAgent
from lxml import etree
from lxml.html.soupparser import fromstring
from parse import search as parse_search
from parse import findall
from w3lib.encoding import html_to_unicode
DEFAULT_ENCODING = 'utf-8'
useragent = UserAgent()
class HTMLResponse(requests.Response):
"""An HTML-enabled :class:`Response <Response>` object.
Same as Requests class:`Response <Response>` object, but with an
intelligent ``.html`` property added.
"""
def __init__(self, *args, **kwargs):
super(HTMLResponse, self).__init__(*args, **kwargs)
self._html = None
@property
def html(self):
if self._html:
return self._html
self._html = HTML(url=self.url, html=self.text, default_encoding=self.encoding)
return self._html
@classmethod
def _from_response(cls, response):
html_r = cls()
html_r.__dict__.update(response.__dict__)
return html_r
class BaseParser:
"""A basic HTML/Element Parser, for Humans."""
def __init__(self, *, element, default_encoding=None, html=None, url):
self.element = element
self.url = url
self.skip_anchors = True
self.default_encoding = default_encoding
self._encoding = None
self._html = html
@property
def html(self):
"""Unicode representation of the HTML content."""
if self._html:
return self._html
else:
return etree.tostring(self.element, encoding='unicode').strip()
@html.setter
def set_html(self, html):
"""Property setter for self.html."""
self._html = html
@property
def encoding(self):
"""The encoding string to be used, extracted from the HTML and
:class:`HTMLResponse <HTMLResponse>` headers.
"""
if self._encoding:
return self._encoding
# Scan meta tags for chaset.
if self._html:
self._encoding = html_to_unicode(self.default_encoding, self.html.encode(DEFAULT_ENCODING))[0]
return self._encoding if self._encoding else self.default_encoding
@property
def pq(self):
"""PyQuery representation of the :class:`Element <Element>` or :class:`HTML <HTML>`."""
return PyQuery(self.element)
@property
def lxml(self):
return fromstring(self.html)
@property
def text(self):
"""The text content of the :class:`Element <Element>` or :class:`HTML <HTML>`."""
return self.pq.text()
@property
def full_text(self):
"""The full text content (including links) of the :class:`Element <Element>` or :class:`HTML <HTML>`.."""
return self.lxml.text_content()
def find(self, selector, first=False, _encoding=None):
"""Given a jQuery selector, returns a list of :class:`Element <Element>` objects.
If ``first`` is ``True``, only returns the first :class:`Element <Element>` found."""
def gen():
for found in self.pq(selector):
yield Element(element=found, url=self.url, default_encoding=_encoding or self.encoding)
c = [g for g in gen()]
if first:
try:
return c[0]
except IndexError:
return None
else:
return c
def xpath(self, selector, first=False, _encoding=None):
"""Given an XPath selector, returns a list of :class:`Element <Element>` objects.
If ``first`` is ``True``, only returns the first :class:`Element <Element>` found."""
c = [Element(element=e, url=self.url, default_encoding=_encoding or self.encoding) for e in self.lxml.xpath(selector)]
if first:
try:
return c[0]
except IndexError:
return None
else:
return c
def search(self, template):
"""Searches the :class:`Element <Element>` for the given parse template."""
return parse_search(template, self.html)
def search_all(self, template):
"""Searches the :class:`Element <Element>` (multiple times) for the given parse
template.
"""
return [r for r in findall(template, self.html)]
@property
def links(self):
"""All found links on page, in as–is form."""
def gen():
for link in self.find('a'):
try:
href = link.attrs['href'].strip()
if not href.startswith('#') and self.skip_anchors and href not in ['javascript:;']:
if href:
yield href
except KeyError:
pass
return set(g for g in gen())
@property
def absolute_links(self):
"""All found links on page, in absolute form."""
def gen():
for link in self.links:
# Parse the link with stdlib.
parsed = urlparse(link)._asdict()
# Appears to be a relative link:
if not parsed['netloc']:
parsed['netloc'] = urlparse(self.base_url).netloc
if not parsed['scheme']:
parsed['scheme'] = urlparse(self.base_url).scheme
# Re-construct URL, with new data.
parsed = (v for v in parsed.values())
href = urlunparse(parsed)
yield href
return set(g for g in gen())
@property
def base_url(self):
"""The base URL for the page. Supports the ``<base>`` tag."""
# Support for <base> tag.
base = self.find('base', first=True)
if base:
return base.attrs['href'].strip()
else:
url = '/'.join(self.url.split('/')[:-1])
if url.endswith('/'):
url = url[:-1]
return url
class Element(BaseParser):
"""An element of HTML."""
def __init__(self, *, element, url, default_encoding):
super(Element, self).__init__(element=element, url=url, default_encoding=default_encoding)
self.element = element
def __repr__(self):
attrs = []
for attr in self.attrs:
attrs.append('{}={}'.format(attr, repr(self.attrs[attr])))
return "<Element {} {}>".format(repr(self.element.tag), ' '.join(attrs))
@property
def attrs(self):
"""Returns a dictionary of the attributes of the class:`Element <Element>`."""
attrs = {k: self.pq.attr[k].strip() for k in self.element.keys()}
# Split class up, as there are ussually many of them:
if 'class' in attrs:
attrs['class'] = tuple(attrs['class'].split())
return attrs
class HTML(BaseParser):
"""An HTML document, ready for parsing."""
def __init__(self, *, url, html, default_encoding=DEFAULT_ENCODING):
super(HTML, self).__init__(
element=PyQuery(html)('html'),
html=html,
url=url,
default_encoding=default_encoding
)
def __repr__(self):
return "<HTML url={}>".format(repr(self.url))
def user_agent(style=None):
"""Returns a random user-agent, if not requested one of a specific
style.
"""
if not style:
return useragent.random
else:
return useragent[style]
class HTMLSession(requests.Session):
"""A consumable session, for cookie persistience and connection pooling,
amongst other things.
"""
def __init__(self, mock_browser=True, *args, **kwargs):
super(HTMLSession, self).__init__(*args, **kwargs)
# Mock a web browser's user agent.
if mock_browser:
self.headers['User-Agent'] = user_agent()
self.hooks = {'response': self._handle_response}
@staticmethod
def _handle_response(response, **kwargs):
"""Requests HTTP Response handler. Attaches .html property to Response
objects.
"""
if not response.encoding:
response.encoding = DEFAULT_ENCODING
return response
def request(self, *args, **kwargs):
# Convert Request object into HTTPRequest object.
r = super(HTMLSession, self).request(*args, **kwargs)
html_r = HTMLResponse._from_response(r)
return html_r
class BrowserHTMLSession(HTMLSession):
"""A web-browser interpreted session (for JavaScript), powered by
`PyPpeteer <https://pypi.python.org/pypi/pyppeteer>`_."""
def __init__(self, *args, **kwargs):
super(BrowserHTMLSession, self).__init__(*args, **kwargs)
def request(self, *args, **kwargs):
# Convert Request object into HTTPRequest object.
r = super(BrowserHTMLSession, self).request(*args, **kwargs)
r._content = self.render(r.url).encode(DEFAULT_ENCODING)
r.encoding = DEFAULT_ENCODING
return r
@staticmethod
def render(source_url):
"""Fully render HTML, JavaScript and all."""
async def _async_render(url):
browser = pyppeteer.launch()
page = await browser.newPage()
await page.goto(url)
content = await page.content()
return content
loop = asyncio.get_event_loop()
content = loop.run_until_complete(_async_render(source_url))
return content
# Backwards compatiblity.
session = HTMLSession()
Session = HTMLSession
BrowserSession = BrowserHTMLSession