Python爬虫总结

1.1 urllib库

示例代码如下：

1.1.1发送get请求

  1. from urllib import request
  3. url = 'https://qiubai-video-web.qiushibaike.com/MK11R93Y1A96C557_hd.mp4'
  5. resp = request.urlopen(url) # 返回 http.client.HTTPResponse object
  7. if resp.getcode() == 200:
  8.     with open('a.mp4', 'wb') as f:
  9.         f.write(resp.read())

1.1.2发送post请求

  1. from urllib import request
  2. from urllib import parse
  4. url = "https://study.163.com/mob/search/independent/v1"
  6. data = {
  7.     "keyword": "王顺子",
  8.     "pageIndex": "1",
  9.     "pageSize": "20",
  10.     "searchType": "0"
  11. }
  13. params_str = parse.urlencode(data)
  15. resp = request.urlopen(url, data=bytes(params_str, encoding='UTF-8'))

1.1.3获取状态码

  1. resp.getcode()

1.1.4获取响应头信息

获取所有头部信息

  1. resp.getheaders()

获取某一字段信息

  1. resp.getheader('Content-Type') # text/html;charset=UTF-8

1.1.5获取响应内容

  1. resp.read()

1.1.6url地址编解码

  1. from urllib import parse
  3. params = {
  4.     'kw': 'python教程',
  5.     'searchType': '1'
  6. }
  8. print(parse.urlencode(params))  # kw=python%E6%95%99%E7%A8%8B&searchType=1
  10. print(parse.parse_qs(params_str))  # {'kw': ['python教程'], 'searchType': ['1']}
  12. print(parse.parse_qsl(params_str))  # [('kw', 'python教程'), ('searchType', '1')]

1.1.7构建请求头

示例代码如下：

  1. from urllib import request
  2. from urllib import parse
  4. url = "http://www.baidu.com"
  6. headers = {
  7.     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36"
  8. }
  10. # 1、构建一个Request对象
  11. req = request.Request(url, headers=headers)
  13. # 2、发送网络请求
  14. resp = request.urlopen(req)
  16. print(resp.read().decode('UTF-8'))

1.1.8处理不受信任的证书

示例代码如下：

  1. from urllib import request
  2. import ssl
  4. url = "https://www.baidu.com"
  6. headers = {
  7.     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36"
  8. }
  10. # 1、构建一个Request对象
  11. req = request.Request(url, headers=headers)
  13. # context = ssl._create_unverified_context() # 忽略证书认证
  14. context = ssl.create_default_context(cafile="charles.pem")  # 指定证书
  16. # 2、发送网络请求
  17. resp = request.urlopen(req, context=context)
  18. # resp = request.urlopen(req, cafile='charles.pem')
  20. print(resp.read().decode('UTF-8'))

1.1.9设置代理

  1. from urllib import request
  3. url = "http://httpbin.org/ip"
  5. proxy = '121.230.209.149:18022' # 免费代理
  6. # proxy = '1924086038:xle4zavg@113.57.97.228:19532' # 私密代理
  8. # 1、创建一个httpHandler处理器对象
  9. proxy_handler = request.ProxyHandler({'http': proxy})
  11. # 2、封装成opener对象
  12. proxy_opener = request.build_opener(proxy_handler)
  14. # 3、发送网络请求
  15. resp = proxy_opener.open(url)
  17. print(resp.read().decode('UTF-8'))

1.1.10验证码处理

12306验证码案例

手动处理cookie

  1. from urllib import request
  2. from urllib import parse
  4. # 1. 下载验证码, 保存cookie
  5. url = "https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&0.7367307074501106"
  6. resp = request.urlopen(url)
  8. # 1.1 保存cookie
  9. headers_str = resp.getheader("Set-Cookie")
  10. headers = headers_str.split(",")
  11. headers_result = []
  12. for header in headers:
  13.     # _passport_session=c1ac8bbf9fc44a0ba06274fdadd588cf2003; Path=/passport
  14.     headers_result.append(header.split(";")[0])
  15. headers_result_str = ";".join(headers_result)
  16. print(headers_result_str)
  18. # 1.2 保存验证码图片
  19. with open("yzm.jpg", "wb") as f:
  20.     f.write(resp.read())
  22. answer = input("请输入验证码答案: ")
  24. # 2. 验证验证码操作
  25. # post
  26. check_url = "https://kyfw.12306.cn/passport/captcha/captcha-check"
  27. data = {
  28.     "answer":    answer,
  29.     "login_site":    "E",
  30.     "rand":    "sjrand"
  31. }
  32. param = parse.urlencode(data)
  33. param_bytes = bytes(param, encoding="utf-8")
  35. check_headers = {
  36.     "Cookie": headers_result_str
  37. }
  39. req = request.Request(check_url, data=param_bytes, headers=check_headers)
  40. result = request.urlopen(req)
  41. print(result.read().decode("utf-8"))

自动处理cookie

  1. from urllib import request
  2. from urllib import parse
  3. from http.cookiejar import CookieJar, MozillaCookieJar
  5. # 1. 下载验证码, 保存cookie
  6. url = "https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&0.7367307074501106"
  9. # 1.1 创建一个处理器对象
  10. # cookie_jar = CookieJar()  # CookieJar只是在内存缓存
  11. cookie_jar = MozillaCookieJar()  # MozillaCookieJar对象有一个save方法,可以把cookie保存到磁盘
  12. cookie_handler = request.HTTPCookieProcessor(cookie_jar)
  15. # 1.2 构造一个opener对象
  16. cookie_opener = request.build_opener(cookie_handler)
  19. # 1.3 opener 发送一个网络请求
  20. resp = cookie_opener.open(url)
  22. # print(cookie_jar)
  24. # 查看cookie
  25. for c in cookie_jar:
  26.     print(c)
  28. # cookie_jar.save("cookie.txt", ignore_discard=True, ignore_expires=True)  # 保存cookie
  29. # print(cookie_jar.load("cookie.txt", ignore_discard=True, ignore_expires=True))  # 加载cookie
  32. # 1.2 保存验证码图片
  33. with open("yzm.jpg", "wb") as f:
  34.     f.write(resp.read())
  36. answer = input("请输入验证码答案: ")
  38. # 2. 验证验证码操作
  39. # post
  40. check_url = "https://kyfw.12306.cn/passport/captcha/captcha-check"
  41. data = {
  42.     "answer":    answer,
  43.     "login_site":    "E",
  44.     "rand":    "sjrand"
  45. }
  46. param = parse.urlencode(data)
  47. param_bytes = bytes(param, encoding="utf-8")
  48. req = request.Request(check_url, data=param_bytes)
  49. result = cookie_opener.open(req)
  50. print(result.read().decode("utf-8"))

账号密码登录自动处理

  1. from urllib import request
  3. user = "itlike"
  4. pwd = "123456"
  6. url = "http://httpbin.org/basic-auth/itlike/123456"
  8. # 1. 创建一个特定的处理器对象
  9. pm = request.HTTPPasswordMgrWithDefaultRealm()
  10. pm.add_password(None, url, user, pwd)
  11. handler = request.HTTPBasicAuthHandler(pm)
  13. # 2. 构建出一个Opener对象
  14. opener = request.build_opener(handler)
  16. # 3. 使用opener对象打开所有需要授权验证的URL
  17. resp = opener.open(url)
  19. print(resp.read().decode("utf-8"))

创建多处理器opner对象

  1. from urllib import request
  2. from http.cookiejar import CookieJar
  4. # http
  5. proxy = "111.177.177.28:9999"
  6. # url = "http://httpbin.org/ip"
  7. url = "https://www.baidu.com"
  9. # 1. 创建一个处理器对象Handler
  10. proxy_handler = request.ProxyHandler({"http": proxy})
  12. cookie_jar = CookieJar()
  13. cookie_handler = request.HTTPCookieProcessor(cookie_jar)
  15. opener = request.build_opener(proxy_handler, cookie_handler)
  17. # 3. opener发送请求(url, Request)
  18. req = request.Request(url, headers={
  19.     "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 11_0 like Mac OS X) AppleWebKit/604.1.38 (KHTML, like Gecko) Version/11.0 Mobile/15A372 Safari/604.1"
  20. })
  21. resp = opener.open(req)
  23. for cookie in cookie_jar:
  24.     print(cookie)
  26. # print(resp.read().decode("utf-8"))

1.1.11弹窗验证授权处理

  1. from urllib import request
  2. import base64
  5. user = "itlike"
  6. pwd = "123456"
  8. result_str = user + ":" + pwd
  10. result = "Basic " + base64.b64encode(bytes(result_str, encoding="utf-8")).decode("utf-8")
  11. # print(result)
  12. #
  13. # exit(0)
  15. url = "http://httpbin.org/basic-auth/itlike/123456"
  17. req = request.Request(url, headers={
  18.     "Authorization": result
  19. })
  21. resp = request.urlopen(req)
  23. print(resp.read().decode("utf-8"))

1.1.12下载进度监听

  1. from urllib import request
  3. url = "https://m801.music.126.net/20210113162434/d250854fc6dd71e3ac7f107df2305932/jdyyaac/0353/055e/565e/4910a621a524e5158013a4ccdad535d9.m4a"
  5. def download_msg(block_num, block_size, total_size):
  6.     # print(block_num, block_size, total_size)
  7.     progress = (block_num + 1) * block_size / total_size
  8.     progress = 1 if progress > 1 else progress
  9.     print(progress)
  11. request.urlretrieve(url, "url_test_video.mp4", reporthook=download_msg)

1.1.12异常处理

  1. from urllib import request
  2. from urllib import error
  3. import socket
  5. try:
  6.     url = "http://localhost/test8.mp4"
  8.     resp = request.urlopen(url)
  10.     print(resp.read().decode("utf-8"))
  11. # except error.HTTPError as he:
  12. #     print("http error", he.code, he.msg)
  13. # except error.URLError as ue:
  14. #     print(ue)
  15. # except socket.timeout as te:
  16. #     print(te)
  17. except Exception as e:
  18.     print(e)

1.2 requests库

1.2.1发送get请求

  1. import requests as rts
  3. # get
  4. # mid = 2081377
  5. url = "https://study.163.com/category/480000003131009"
  7. # full_url = url + "?" + "mid=2081377"
  9. pm = {
  10.     "mid": "2081377"
  11. }
  13. # pm = [
  14. #     ("mid", "2081377")
  15. # ]
  17. # rts.request("get", url)
  18. resp = rts.get(url, params=pm)

1.2.2发送post请求

  1. import requests as rts
  3. # post
  4. url = "https://study.163.com/mob/search/independent/v1"
  6. data_dic = {
  7.     "keyword": "python",
  8.     "pageIndex": "1",
  9.     "pageSize": "20",
  10.     "searchType": "0"
  11. }
  13. # resp = rts.request("post", url, data=data_dic, verify=False)
  14. cert = r'charles.pem'
  15. # resp = rts.post(url, json=data_dic, verify=False)  # json参数看post请求携带的参数是json字符串还是使用&符拼接的方式
  16. resp = rts.post(url, data=data_dic, verify=cert)  # verify可以解决证书验证的问题，可以忽略验证或指定证书路径

这里需要注意的是，发送post请求是使用第三方库requests版本位2.25.1时，证书验证verify字段设置不起作用，需要降低版本库。使用requests2.7版本可以解决证书验证的问题。

1.2.3获取状态码

  1. resp.status_code

1.2.4获取响应头信息

  1. print(resp.reason)  # ok
  2. print(resp.ok)  # True
  3. print(resp.headers) # 获取响应头信息

1.2.5获取响应体信息

  1. # print(resp.raw)
  2. # print(type(resp.raw))  # urllib3.response.HTTPResponse
  3. print(resp.content)  # 原始字符串，未编码
  4. print(resp.text)  # 经过编码

1.2.6获取响应编码

  1. print(resp.encoding)  # UTF-8
  2. resp.encoding = 'UTF-8'  # 指定编码

1.2.7构建请求头

  1. import requests as rts
  3. url = "https://www.baidu.com"
  5. headers = {
  6.     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36"
  7. }
  9. resp = rts.get(url, verify=False, headers=headers)

1.2.8处理不受信任的证书

  1. import requests as rts
  3. url = "https://www.1234.com"
  5. resp = rts.get(url, verify=False) # 忽略证书验证
  6. print(resp.text)

1.2.9设置代理

  1. import requests as rts
  3. url = "http://httpbin.org/ip"
  5. proxy = {
  6.     # "http": "110.243.9.204:9999"
  7.     "http": "http://1924086038:xle4zavg@140.250.153.124:21160"  # 私密代理
  8. }
  10. resp = rts.get(url, proxies=proxy)
  12. print(resp.text)

1.2.10验证码处理

手动处理cookies

  1. import requests as rts
  3. yzm_url = "https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&0.7367307074501106"
  5. yzm_resp = rts.get(yzm_url)
  7. cookie = yzm_resp.cookies
  9. with open("yzm.jpg", "wb") as f:
  10.     f.write(yzm_resp.content)
  12. # 回答验证码
  13. answer = input("请输入验证码答案: ")
  14. check_url = "https://kyfw.12306.cn/passport/captcha/captcha-check"
  16. data = {
  17.     "answer":    answer,
  18.     "login_site":    "E",
  19.     "rand":    "sjrand"
  20. }
  22. check_resp = rts.post(check_url, data=data, cookies=cookie)
  24. print(check_resp.text)

自动处理cookie

  1. import requests as rts
  3. session = rts.Session()
  5. yzm_url = "https://kyfw.12306.cn/passport/captcha/captcha-image?login_site=E&module=login&rand=sjrand&0.7367307074501106"
  7. yzm_resp = session.get(yzm_url)
  9. # cookie = yzm_resp.cookies
  10. with open("yzm.jpg", "wb") as f:
  11.     f.write(yzm_resp.content)
  14. # 回答验证码
  15. answer = input("请输入验证码答案: ")
  16. check_url = "https://kyfw.12306.cn/passport/captcha/captcha-check"
  18. data = {
  19.     "answer":    answer,
  20.     "login_site":    "E",
  21.     "rand":    "sjrand"
  22. }
  24. # check_resp = rts.post(check_url, data=data, cookies=cookie)
  25. check_resp = session.post(check_url, data=data)
  27. print(check_resp.text)

1.2.11弹窗验证授权处理

  1. import requests as rts
  3. url = "http://httpbin.org/basic-auth/itlike/123456"
  5. resp = rts.get(url, auth=("itlike", "123456"))
  7. print(resp.text)

1.2.12流式下载

按字节读取

  1. import requests as rts
  3. url = "https://qiubai-video-web.qiushibaike.com/B1E3Q8C4514VKRJ3_hd.mp4"
  5. resp = rts.get(url, stream=True, verify=False, headers={
  6.     "Accept-Encoding": ""
  7. })
  9. print(len(resp.content))
  12. for chunk in resp.iter_content(1024):
  13.     print(chunk)

按行读取

  1. import requests as rts
  3. url = "https://study.163.com/category/480000003131009"
  5. pm = [
  6.     ("mid", "2081377")
  7. ]
  9. resp = rts.get(url, params=pm, verify=False, stream=True)
  11. # print(resp.raw.read(30000))
  12. # print(resp.text)
  14. with open("163.html", "w", encoding="utf-8") as f:
  15.     for chunk in resp.iter_lines():
  16.         f.write(chunk.decode("utf-8"))
  17.         # print(chunk.decode("utf-8"))
  18.         # stop = input("请问是否停止(y/n)")
  19.         # if stop == "y":
  20.         #     break

1.2.14下载进度监听

  1. import requests as rts
  3. url = "https://qiubai-video-web.qiushibaike.com/B1E3Q8C4514VKRJ3_hd.mp4"
  5. resp = rts.get(url, verify=False, stream=True)
  7. print(resp.headers)
  9. # 1. 先来获取文件的总大小
  10. total_size = int(resp.headers["Content-Length"])
  12. # 2. 计算当前已经下载的内容大小
  13. current_size = 0
  14. with open("gaoxiao.mp4", "wb") as f:
  15.     for chunk in resp.iter_content(1024*10):
  16.         f.write(chunk)
  17.         current_size += len(chunk)
  18.         print(current_size / total_size)

1.2.15异常处理

  1. import requests as rts
  3. url = "https://study.163.com/category/480000003131009"
  5. pm = [
  6.     ("mid", "2081377")
  7. ]
  9. try:
  10.     # resp = rts.get(url, params=pm, timeout=0.001)
  11.     resp = rts.get(url, params=pm, timeout=(0.001, 2))
  13.     print(resp.ok)
  14. except rts.exceptions.ConnectTimeout as cte:
  15.     print("连接异常-超时", cte)
  16. except rts.exceptions.ReadTimeout as rte:
  17.     print("读取异常-超时", rte)

1.3 数据解析

1.3.1json字符串解析

  1. import json
  3. json_str = '{"status":{"code":0,"message":""},"extraInfo":null}'
  5. result = json.loads(json_str)
  7. print(type(result))  # <class 'dict'>
  9. print(result)  # {'status': {'code': 0, 'message': ''}, 'extraInfo': None}