python爬虫----使用httpx

使用httpx

1、背景

使用urllib和requests库已经可以爬取绝大部分网站的数据，但对于某些网站任然无能为力，因为这些网站强制使用HTTP/2.0协议访问，这时使用urllib和requests是无法爬取数据的，因为这两个库只支持HTTP/1.1，这个时候就需要用到httpx库了

2、基本使用

1、安装

# 这样安装不支持HTTP/2.0 
pip install httpx
# 安装httpx并支持HTTP/2.0
pip install 'httpx[http2]'
1
2
3
4

2、基本使用

# 导入模块
import httpx
# 发送get请求
r = httpx.get('https://www.httpbin.org/get/')
# 输出响应状态码
print(r.status_code)
# 输出请求头
print(r.headers)
# 输出响应体
print(r.text)

# 说明：httpx默认不会开启对HTTP/2.0的支持，默认使用HTTP/1.1，需要手动声明一次才能使用
import httpx
client = httpx.Client(http2=True)
r = client.get('https://scrape.center/')
print(r.text)

# httpx库与requests库的很多方式类似
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

3、Client 对象

1、使用

# 官方推荐使用with as 语句
# 案例1
import httpx
with httpx.Client() as client:
    r = client.get('https://www.httpbin.org/get')
    print(r)

# 案例2：与案例1等价
import httpx
client = httpx.Client()
try:
    r = client.get('https://www.htpbin.org/get')
finally:
    client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2、指定参数

import httpx
url = "http://www.httpbin.org/headers"
headers = {"User-Agent":"my-app/1.0.0.1"}
with httpx.Client(headers=headers) as client:
    r = client.get(url)
    print(r.json()['headers']['User-Agent'])
    print(r.http_version)
1
2
3
4
5
6
7

3、支持异步请求
httpx支持异步客户端请求（AsyncClient），支持Python的async请求模式

import httpx
import asyncio
async def fetch(url):
    async with httpx.AsyncClientt(http2=True) as client:
        r = await client.get(url)
        print(r.text)
if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(fetch('https://www.httpbin.org/get'))
1
2
3
4
5
6
7
8

3、基本爬虫案例实战

1、准备工作

1、安装Python环境
2、了解Python多进程的基本原理
3、了解Python HTTP请求库requests的基本用法
4、了解正则表达式的用法和Python中正则表达式库re的基本用法

2、爬取目标

爬取基本静态网站：电影网站
目标网址：https://movie.douban.com/top250

需完成目标：
1）利用requests爬取这个站点每一页的电影列表，顺着列表再爬取每个电影的详情页
2）用正则表达式提取每部电影的名称、封面、类别、上映时间、评分、剧情简介等内容
3）把爬取的内容保存为JSON文本文件
4）使用多进程实现爬取加速

1、分析列表页
观察列表页结构和翻页规则
分析网页结构
确认爬取目标

2、实现列表页爬取
遍历所有页码，构造10页的索引页URL
从每个索引页分析提取出每个电影的详情页URL

requests请求

# 用来请求网页
import requests
# 用来输出信息
import logging
# 用来实现正则表达式解析
import re
# 用来做URL拼接
from urllib.parse import urljoin

# 定义日志输出级别和输出格式
logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s - %(levelname)s: %(message)s")
BASE_URL = 'https://movie.douban.com/top250'

# 页码总数量
TOTAL_PAGE = 10

# 定义页面爬取方法
def scrape_page(url):
    logging.info('scraping %s...', url)
    try:
        r = requests.get(url)
        # 判断状态码是否为200，如果是，直接返回页面的HTML代码；如果不是，则输出错误日志信息
        if r.status_code == 200:
            return r.text
        logging.error('get invalid status code %s while scraping %s',
                      r.status_code, url)
    except requests.RequestException:
        # 如果出现爬取异常，输出对应的错误日志信息
        logging.error('error occurred while scraping %s', url,
                      exc_info=True)

# 定义列表爬取方法：这个方法会接收一个page参数，即列表页的页码，在方法里面实现列表页的URL拼接，然后调用scrape_page方法爬取即可
def scrape_index(page):
    index_url = f'{BASE_URL}/page/{page}'
    return scrape_page(index_url)

# 解析列表页方法
def parse_index(html):
    pattern = re.compile('<a.*?href="(.*?)".*?class="name">')
    items = re.findall(pattern, html)
    if not items:
        return []
    for item in items:
        # 拼接完整URL
        detail_url = urljoin(BASE_URL, item)
        logging.info('get detail url %s', detail_url)
        yield detail_url

# 主函数
def main():
    for page in range(1, TOTAL_PAGE+1):
        index_html = scrape_index(page)
        detail_urls = parse_index(index_html)
        logging.info('detail urls %s', list(detail_urls))

# 函数开始位置
if __name__ == "__main__":
    main()
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

httx请求

# 用来请求网页
import httpx
# 用来输出信息
import logging
# 用来实现正则表达式解析
import re
# 用来做URL拼接
from urllib.parse import urljoin

# 定义日志输出级别和输出格式
logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s - %(levelname)s: %(message)s")
BASE_URL = 'https://movie.douban.com/top250'

# 页码总数量
TOTAL_PAGE = 10

# 定义页面爬取方法
def scrape_page(url):
    logging.info('scraping %s...', url)
    try:
        r = httpx.get(url)
        # 判断状态码是否为200，如果是，直接返回页面的HTML代码；如果不是，则输出错误日志信息
        if r.status_code == 200:
            return r.text
        logging.error('get invalid status code %s while scraping %s',
                      r.status_code, url)
    except httpx.RequestError:
        # 如果出现爬取异常，输出对应的错误日志信息
        logging.error('error occurred while scraping %s', url,
                      exc_info=True)

# 定义列表爬取方法：这个方法会接收一个page参数，即列表页的页码，在方法里面实现列表页的URL拼接，然后调用scrape_page方法爬取即可
def scrape_index(page):
    index_url = f'{BASE_URL}/page/{page}'
    return scrape_page(index_url)

# 解析列表页方法
def parse_index(html):
    pattern = re.compile('<a.*?href="(.*?)".*?class="name">')
    items = re.findall(pattern, html)
    if not items:
        return []
    for item in items:
        # 拼接完整URL
        detail_url = urljoin(BASE_URL, item)
        logging.info('get detail url %s', detail_url)
        yield detail_url

# 主函数
def main():
    for page in range(1, TOTAL_PAGE+1):
        index_html = scrape_index(page)
        detail_urls = parse_index(index_html)
        logging.info('detail urls %s', list(detail_urls))

# 函数开始位置
if __name__ == "__main__":
    main()

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