常见 HTTP CODE

image.png

Requests 模拟登录知乎

• 先输入用户名密码进行登录测试

image.png
image.png
image.png
image.png

• 发现主要发送了 3 个请求，一次 sing_in 和两次 captcha?lang=en

• sing_in：登录请求，Request URL 为：https://www.zhihu.com/api/v3/oauth/sign_in
• captcha?lang=en：获取验证码，Request URL 为：https://www.zhihu.com/api/v3/oauth/captcha?lang=en

• Request Payload 所有参数

------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="client_id"

c3cef7c66a1843f8b3a9e6a1e3160e20
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="grant_type"

password
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="timestamp"

1528552667566
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="source"

com.zhihu.web
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="signature"

a66dbcd96828a60de1120e4419cd1eb8fa79145f
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="username"

123@163.com
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="password"

123456
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="captcha"


------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="lang"

en
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="ref_source"

homepage
------WebKitFormBoundarypgxZVHAByOSf1fnG
Content-Disposition: form-data; name="utm_source"


------WebKitFormBoundarypgxZVHAByOSf1fnG--

分析发现，sing_in 的时候下面几个请求参数是需要传递的

• timestamp：当前时间戳
• signature：签名认证
• username：用户名
• password：密码
• captcha：验证码

其他的参数全部是固定的，比较不容易搞定的就是 signature 和 captcha

• 如何获取 signature
  在网页源码里找 js 文件，有这样一段代码，就是生成这个 signature 的

function (e, t, n) {
    "use strict";
    function r(e, t) {
        var n = Date.now(), r = new a.a("SHA-1", "TEXT");
        return r.setHMACKey("d1b964811afb40118a12068ff74a12f4", "TEXT"), r.update(e), r.update(i.a), r.update("com.zhihu.web"), r.update(String(n)), c({
            clientId: i.a,
            grantType: e,
            timestamp: n,
            source: "com.zhihu.web",
            signature: r.getHMAC("HEX")
        }, t)
    }

分析发现主要使用 hmac、sha1 进行了加密，可以用 Python 实现同样算法

def get_signature(login_time):
    """
        生成 signature
    """
    h = hmac.new(key='d1b964811afb40118a12068ff74a12f4'.encode('utf-8'), digestmod=sha1)
    client_id = 'c3cef7c66a1843f8b3a9e6a1e3160e20'
    grant_type = 'password'
    timestamp = login_time
    source = 'com.zhihu.web'
    h.update((grant_type + client_id + source + timestamp).encode('utf-8'))
    return h.hexdigest()

• 如何获取 captcha

image.png
image.png

• 可以发现，请求了两次 captcha?lang=en，为什么是两次呢？原因是有的时候是不需要验证码的，直接输入用户名密码就可以完成登录，这样最好，但是有的时候是会出现验证码的，第一次请求 captcha?lang=en 返回结果是 {“show_captcha”:true} 为 ‘true’ 代表需要验证码，如果第一次返回 {“show_captcha”:false} 则代表不需要图形验证码，也就不会有第二次请求真正的验证的了
• 这里要说明一下，请求 Request URL:https://www.zhihu.com/api/v3/oauth/captcha?lang=en 最后的 lang=en 代表英文验证码，如果 lang=cn 就会出现倒立的中文验证码，所以发送请求的时候要发送能够获取英文验证码的请求

获取图形验证码

def get_captcha(headers):
    """
        获取登录验证码
    Args:
        headers: 请求头信息
    Returns:
        captcha: 返回手动输入的验证码
    """
    response = session.get('https://www.zhihu.com/api/v3/oauth/captcha?lang=en', headers=headers)
    result = re.findall('"show_captcha":(\w+)', response.text)
    if result[0] == 'false':
        return ''
    else:
        response = session.get('https://www.zhihu.com/api/v3/oauth/captcha?lang=en', headers=headers)
        show_captcha = json.loads(response.text)['img_base64']
        with open('captcha.jpg', 'wb') as f:
            f.write(base64.b64decode(show_captcha.encode('utf-8')))
        image = Image.open('captcha.jpg')
        image.show()
        image.close()
        captcha = input('请输入验证码：')
        return captcha

• 两个难点解决了，其他问题就都好处理了
  完整代码如下

import re
import time
import http.cookiejar
import hmac
from hashlib import sha1
import base64
import json

import requests
from PIL import Image


session = requests.Session()
session.cookies = http.cookiejar.LWPCookieJar(filename='cookies.txt')
try:
    session.cookies.load(ignore_discard=True)
except Exception as e:
    print('cookie 未能加载')


headers = {
    'authorization': 'oauth c3cef7c66a1843f8b3a9e6a1e3160e20',
    'host': 'www.zhihu.com',
    'referer': 'https://www.zhihu.com/',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36',
}


def is_login():
    """
        判断是否已经登录
    """
    response = session.get('https://www.zhihu.com', headers=headers, allow_redirects=False)
    if response.status_code != 200:
        return False
    else:
        return True


def get_xsrf_d_c0():
    """
        获取 _xsrf 和 d_c0 的值
    """
    response = session.get('https://www.zhihu.com/signup', headers=headers)
    return response.cookies.get('_xsrf'), response.cookies.get('d_c0')


def get_signature(login_time):
    """
        生成 signature
    """
    h = hmac.new(key='d1b964811afb40118a12068ff74a12f4'.encode('utf-8'), digestmod=sha1)
    client_id = 'c3cef7c66a1843f8b3a9e6a1e3160e20'
    grant_type = 'password'
    timestamp = login_time
    source = 'com.zhihu.web'
    h.update((grant_type + client_id + source + timestamp).encode('utf-8'))
    return h.hexdigest()


def get_captcha(headers):
    """
        获取登录验证码
    Args:
        headers: 请求头信息
    Returns:
        captcha: 返回手动输入的验证码
    """
    response = session.get('https://www.zhihu.com/api/v3/oauth/captcha?lang=en', headers=headers)
    result = re.findall('"show_captcha":(\w+)', response.text)
    if result[0] == 'false':
        return ''
    else:
        response = session.get('https://www.zhihu.com/api/v3/oauth/captcha?lang=en', headers=headers)
        show_captcha = json.loads(response.text)['img_base64']
        with open('captcha.jpg', 'wb') as f:
            f.write(base64.b64decode(show_captcha.encode('utf-8')))
        image = Image.open('captcha.jpg')
        image.show()
        image.close()
        captcha = input('请输入验证码：')
        return captcha


def zhihu_login(user, password):
    """
        知乎登录
    Args:
        user: 用户名
        password: 密码
    Returns:
        response: 响应信息
    """
    x_xsrftoken, x_udid = get_xsrf_d_c0()
    headers.update({
        # 'x-udid': x_udid,
        'x-xsrftoken': x_xsrftoken,
    })
    login_time = str(int(time.time() * 1000))
    post_url = 'https://www.zhihu.com/api/v3/oauth/sign_in'
    post_data = {
        'client_id': 'c3cef7c66a1843f8b3a9e6a1e3160e20',
        'grant_type': 'password',
        'timestamp': login_time,
        'source': 'com.zhihu.web',
        'signature': get_signature(login_time),
        'username': user,
        'password': password,
        'captcha': get_captcha(headers),
        'lang': 'en',
        'ref_source': 'homepage',
        'utm_source': '',
    }
    response = session.post(post_url, data=post_data, headers=headers)
    session.cookies.save()
    return response


def get_index():
    """
        请求知乎首页
    Returns:
        返回首页 HTML 源码
    """
    response = session.get('https://www.zhihu.com', headers=headers)
    return response.text


if __name__ == '__main__':
    USER = 'username'  # 用户名
    PASSWORD = 'password'  # 密码
    if not is_login():
        print('正在登录...')
        zhihu_login(USER, PASSWORD)
        html = get_index()
        print(html)
    else:
        print('已登录...')

将 USER 和 PASSWORD 换成自己的用户名和密码就可以登录了

Scrapy 模拟知乎登录1

• 新建 知乎 爬虫

scrapy genspider zhihu www.zhihu.com

image.png
image.png

• 整体思路：模拟登录过程和之前通过 requests 模拟登录时差不多的，不过在 scrapy 中爬虫第一个请求会从 start_urls 列表中开始爬取，并且是运行爬虫后自动开始爬取，而自动爬取是通过 start_requests 方法来实现的，所以要先重写 start_requests 方法，第一个请求改成不从 start_urls 列表中开始，而是从知乎的登录网址 https://www.zhihu.com/signup 开始，登录成功后，再调用 start_urls 列表中的起始 URL 进行真正的数据爬取，然后调用 parse 方法进行数据解析

• 知乎爬虫 模拟登录 完整代码 zhihu.py

# ArticleSpider/spiders/zhihu.py

# -*- coding: utf-8 -*-
import time
import re
import hmac
from hashlib import sha1
import base64
import json
from PIL import Image
import scrapy
from ArticleSpider.settings import ZHIHU_USER, ZHIHU_PASSWORD


class ZhihuSpider(scrapy.Spider):
    name = 'zhihu_login'
    allowed_domains = ['www.zhihu.com']
    start_urls = ['https://www.zhihu.com/topic']
    headers = {
        'authorization': 'oauth c3cef7c66a1843f8b3a9e6a1e3160e20',
        'HOST': 'www.zhihu.com',
        'Referer': 'https://www.zhihu.com/',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36',
    }

    def start_requests(self):
        """
            重写爬虫入口 start_requests，以完成模拟登录知乎
        """
        # 先请求登录页面，从登录页面获取登录所需数据，再进行登录
        return [
            scrapy.Request('https://www.zhihu.com/signup', headers=self.headers, callback=self.get_captcha1)
        ]

    def get_signature(self, time_str):
        """
            生成 signature
        """
        h = hmac.new(key='d1b964811afb40118a12068ff74a12f4'.encode('utf-8'), digestmod=sha1)
        client_id = 'c3cef7c66a1843f8b3a9e6a1e3160e20'
        grant_type = 'password'
        timestamp = time_str
        source = 'com.zhihu.web'
        h.update((grant_type + client_id + source + timestamp).encode('utf-8'))
        return h.hexdigest()

    def get_captcha1(self, response):
        """
            第一次请求获取登录验证码
        """
        pattern = re.compile('_xsrf=(.*?);')
        cookies = response.headers.getlist('Set-Cookie')
        xsrf = re.findall(pattern, str(cookies))
        xsrf = xsrf[0] if xsrf else ''
        self.headers.update({'x-xsrftoken': xsrf})
        time_str = str(int(time.time() * 1000))
        signature = self.get_signature(time_str)

        return [
            scrapy.Request(
                url='https://www.zhihu.com/api/v3/oauth/captcha?lang=en',
                headers=self.headers,
                callback=self.get_captcha2,
                meta={'time_str': time_str, 'signature': signature},
                dont_filter=True
            )
        ]

    def get_captcha2(self, response):
        """
            第二次请求获取登录验证码
            如果判断不需要验证码，则直接进行登录
        """
        result = re.findall('"show_captcha":(\w+)', response.text)
        formdata = {
            'client_id': 'c3cef7c66a1843f8b3a9e6a1e3160e20',
            'grant_type': 'password',
            'timestamp': response.meta['time_str'],
            'source': 'com.zhihu.web',
            'signature': response.meta['signature'],
            'username': ZHIHU_USER,
            'password': ZHIHU_PASSWORD,
            'captcha': '',
            'lang': 'en',
            'ref_source': 'homepage',
            'utm_source': '',
        }

        if result[0] == 'false':
            # 不需要输入验证码，直接进行登录
            return [
                scrapy.FormRequest(  # FormRequest 可以实现 POST 请求完成表单提交
                    url='https://www.zhihu.com/api/v3/oauth/sign_in',
                    formdata=formdata,
                    headers=self.headers,
                    callback=self.check_login,
                    dont_filter=True
                )
            ]
        else:
            # 需要输入验证码，再次请求获取验证码
            # 请求回调 parse_captcha 方法进行验证码解析
            return [
                scrapy.Request(
                    url='https://www.zhihu.com/api/v3/oauth/captcha?lang=en',
                    headers=self.headers,
                    callback=self.parse_captcha,
                    meta={'formdata': formdata},
                    dont_filter=True
                )
            ]

    def parse_captcha(self, response):
        """
            解析验证码，并将验证码图片存储到本地
            需要手动输入验证码后继续执行代码完成登录
        """
        show_captcha = json.loads(response.text)['img_base64']
        with open('captcha.jpg', 'wb') as f:
            f.write(base64.b64decode(show_captcha.encode('utf-8')))
        image = Image.open('captcha.jpg')
        image.show()
        image.close()
        captcha = input('请输入验证码：')
        formdata = response.meta['formdata']
        formdata['captcha'] = captcha

        return [
            scrapy.FormRequest(  # FormRequest 可以实现 POST 请求完成表单提交
                url='https://www.zhihu.com/api/v3/oauth/sign_in',
                formdata=formdata,
                headers=self.headers,
                callback=self.check_login,
                dont_filter=True
            )
        ]

    def check_login(self, response):
        """
            验证服务器返回数据，判断是否登录成功
            如果登录成功，则进行进一步的数据爬取
        """
        if response.status == 200 or response.status == 201:
            # 判断登录成功后，开始进行数据爬取，相当于延后了
            # start_urls 中的 URL 爬取，先完成上面的登录过程
            # 准备工作做完，在这里再开始进行真正的数据爬取
            for url in self.start_urls:
                # 不写 callback 参数默认回调时调用 parse 方法
                yield scrapy.Request(url, headers=self.headers, dont_filter=True)

    def parse(self, response):
        """
            数据解析
        """
        print(response.text)
        pass

• 还需要在 settings.py 中配置用户名和密码

# ArticleSpider/settings.py

# 知乎 配置
ZHIHU_USER = 'username'
ZHIHU_PASSWORD = 'password'

Scrapy 模拟知乎登录2

成功模拟登录以后，无法正常访问知乎首页 https://www.zhihu.com/，开始以为模拟登录有问题，就重新分析了一下登录请求过程，重写了登录逻辑，发现依旧不能访问首页，模拟登录后其他页面是可以访问的（如知乎话题页：https://www.zhihu.com/topic）。不过写都写了，代码贴出来，这次逻辑更加清晰了。不过经过实际测试：之前的模拟登录方式从来没遇到过需要输入验证码才能登录的情况，这次的模拟登录每次都需要验证码才能登录

• 不需要验证码情况下的一次完整登录请求

  image.png

image.png
image.png
image.png
image.png
image.png

• 需要验证码的情况下一次完整的登录请求

image.png
image.png
image.png
image.png
image.png
image.png
image.png
image.png
image.png
image.png

• 完整的 知乎登录 代码

# ArticleSpider/spiders/zhihu_login.py

# -*- coding: utf-8 -*-
import time
import hmac
from hashlib import sha1
import base64
import json
from PIL import Image
import scrapy
from ArticleSpider.settings import ZHIHU_USER, ZHIHU_PASSWORD


class ZhihuSpider(scrapy.Spider):
    name = 'zhihu_login'
    allowed_domains = ['www.zhihu.com']
    start_urls = ['https://www.zhihu.com/topic']
    headers = {
        'authorization': 'oauth c3cef7c66a1843f8b3a9e6a1e3160e20',
        'Host': 'www.zhihu.com',
        'Referer': 'https://www.zhihu.com/',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36',
    }
    time_str = str(int(time.time() * 1000))
    formdata = {
        'client_id': 'c3cef7c66a1843f8b3a9e6a1e3160e20',
        'grant_type': 'password',
        'timestamp': time_str,
        'source': 'com.zhihu.web',
        'signature': '',
        'username': ZHIHU_USER,
        'password': ZHIHU_PASSWORD,
        'captcha': '',
        'lang': 'en',
        'ref_source': 'homepage',
        'utm_source': '',
    }

    def start_requests(self):
        """
            重写爬虫入口 start_requests，以完成模拟登录知乎
        """
        # 先请求登录页面，从登录页面获取登录所需数据，再进行登录
        return [
            scrapy.Request(url='https://www.zhihu.com/api/v3/oauth/captcha?lang=en',
                           headers=self.headers,
                           callback=self.is_show_captcha)
        ]

    def get_signature(self, time_str):
        """
            生成 signature
        """
        h = hmac.new(key='d1b964811afb40118a12068ff74a12f4'.encode('utf-8'), digestmod=sha1)
        client_id = 'c3cef7c66a1843f8b3a9e6a1e3160e20'
        grant_type = 'password'
        timestamp = time_str
        source = 'com.zhihu.web'
        h.update((grant_type + client_id + source + timestamp).encode('utf-8'))
        return h.hexdigest()

    def is_show_captcha(self, response):
        """
            判断是否需要验证码登录
        """
        show_captcha = json.loads(response.text)['show_captcha']
        if show_captcha:
            print('需要获取登录验证码')
            return [
                scrapy.Request(url='https://www.zhihu.com/api/v3/oauth/captcha?lang=en',
                               headers=self.headers,
                               callback=self.parse_captcha,
                               method='PUT',
                               dont_filter=True)
            ]
        else:
            print('无需获取登录验证码，直接登录')
            formdata = self.formdata
            formdata['signature'] = self.get_signature(self.time_str)
            return [
                scrapy.FormRequest(  # FormRequest 可以实现 POST 请求完成表单提交
                    url='https://www.zhihu.com/api/v3/oauth/sign_in',
                    formdata=formdata,
                    headers=self.headers,
                    callback=self.check_login,
                    dont_filter=True)
            ]

    def parse_captcha(self, response):
        """
            解析验证码，并将验证码图片存储到本地
            需要手动输入验证码后继续执行代码完成登录
        """
        try:
            show_captcha = json.loads(response.text)['img_base64']
            with open('captcha.jpg', 'wb') as f:
                f.write(base64.b64decode(show_captcha.encode('utf-8')))
            image = Image.open('captcha.jpg')
            image.show()
            # image.close()
            captcha = input('请输入验证码：')
            formdata = {'input_text': captcha}
        except ValueError:
            print('img_base64 获取失败')
        except Exception as e:
            formdata = {}
            print(f'程序出错：{e}')

        return [
            scrapy.FormRequest(  # FormRequest 可以实现 POST 请求完成表单提交
                url='https://www.zhihu.com/api/v3/oauth/captcha?lang=en',
                formdata=formdata,
                headers=self.headers,
                callback=self.captcha_login,
                dont_filter=True)
        ]

    def captcha_login(self, response):
        """
            验证码登录
        """
        try:
            captcha_success = json.loads(response.text)['success']
        except ValueError:
            print('验证码验证失败')
        if captcha_success:
            print('验证码验证成功，正在登录...')
            formdata = self.formdata
            formdata['signature'] = self.get_signature(self.time_str)
            print(formdata)
            return [
                scrapy.FormRequest(  # FormRequest 可以实现 POST 请求完成表单提交
                    url='https://www.zhihu.com/api/v3/oauth/sign_in',
                    formdata=formdata,
                    headers=self.headers,
                    callback=self.check_login,
                    dont_filter=True)
            ]

    def check_login(self, response):
        """
            验证服务器返回数据，判断是否登录成功
            如果登录成功，则进行进一步的数据爬取
        """
        print(response)
        if response.status == 200 or response.status == 201:
            # 判断登录成功后，开始进行数据爬取，相当于延后了
            # start_urls 中的 URL 爬取，先完成上面的登录过程
            # 准备工作做完，在这里再开始进行真正的数据爬取
            for url in self.start_urls:
                # 不写 callback 参数默认回调时调用 parse 方法
                yield scrapy.Request(url, headers=self.headers, dont_filter=True)

    def parse(self, response):
        """
            数据解析
        """
        print(response.text)

爬取知乎问题以及回答

• 对页面进行分析

知乎首页随意点击一个问题进去

image.png

链接如：https://www.zhihu.com/question/280139065/answer/415927966 只显示了当前问题下面的一个回答

image.png

点击查看全部回答，发现URL链接改变了：https://www.zhihu.com/question/280139065

image.png

综上，两个链接进行对比：
单个回答：https://www.zhihu.com/question/280139065/answer/415927966
全部回答：https://www.zhihu.com/question/280139065
可以考虑访问类似：https://www.zhihu.com/question/280139065 这样的 URL 链接进行爬取

• 明确问题需要爬取的信息
  问题详情页面需要提取的信息

image.png

• 明确回答页面需要提取的信息
  分析问题的全部回答页面，发现有 Ajax 请求，复制链接地址，进行访问

image.png

返回的 JSON 数据，刚好是需要采集的回答内容

image.png
image.png
image.png
image.png
image.png

• 设计数据库表

知乎问题表：zhihu_question

image.png

知乎问题回答表：zhihu_answer

image.png

• 定义 ITEM

# ArticleSpider/items.py

class ZhihuQuestionItem(scrapy.Item):
    """
        知乎问题 Item
    """
    # 问题 id
    zhihu_id = scrapy.Field()
    # 所属话题
    topics = scrapy.Field()
    # 问题 URL
    url = scrapy.Field()
    # 问题标题
    title = scrapy.Field(
        input_processor=MapCompose(lambda x: x.strip())
    )
    # 问题内容
    content = scrapy.Field()
    # 问题回答数量
    answer_num = scrapy.Field()
    # 问题评论数
    comments_num = scrapy.Field()
    # 问题关注数
    watch_user_num = scrapy.Field()
    # 问题点击数(浏览数)
    click_num = scrapy.Field()
    # 问题创建时间
    create_time = scrapy.Field()
    # 问题更新时间
    update_time = scrapy.Field()
    # 问题爬取时间
    crawl_time = scrapy.Field()


class ZhihuAnswerItem(scrapy.Item):
    """
        知乎问题回答 Item
    """
    # 回答 id
    zhihu_id = scrapy.Field()
    # 回答 URL
    url = scrapy.Field()
    # 所属问题 id
    question_id = scrapy.Field()
    # 用户 id
    author_id = scrapy.Field()
    # 用户名
    author_name = scrapy.Field()
    # 回答内容
    content = scrapy.Field()
    # 赞同数
    praise_num = scrapy.Field()
    # 评论数
    comments_num = scrapy.Field()
    # 回答创建时间
    create_time = scrapy.Field()
    # 回答修改时间
    update_time = scrapy.Field()
    # 回答爬取时间
    crawl_time = scrapy.Field()

• 编写爬虫

# ArticleSpider/spiders/zhihu_question.py

# -*- coding: utf-8 -*-
import re
import json
import random
import time
from urllib.parse import urljoin
import scrapy
from scrapy.loader import ItemLoader
from ArticleSpider.items import ZhihuQuestionItem, ZhihuAnswerItem


class ZhihuQuestionSpider(scrapy.Spider):
    name = 'zhihu_question'
    allowed_domains = ['www.zhihu.com']
    # start_urls = ['https://www.zhihu.com/explore']  # 重写了爬虫入口 start_requests，所以 start_urls 也就没必要了
    # question 的第一页 answer 请求 URL
    start_answer_url = 'https://www.zhihu.com/api/v4/questions/{question}/answers?sort_by=default&include=data%5B%2A%5D.is_normal%2Cadmin_closed_comment%2Creward_info%2Cis_collapsed%2Cannotation_action%2Cannotation_detail%2Ccollapse_reason%2Cis_sticky%2Ccollapsed_by%2Csuggest_edit%2Ccomment_count%2Ccan_comment%2Ccontent%2Ceditable_content%2Cvoteup_count%2Creshipment_settings%2Ccomment_permission%2Ccreated_time%2Cupdated_time%2Creview_info%2Crelevant_info%2Cquestion%2Cexcerpt%2Crelationship.is_authorized%2Cis_author%2Cvoting%2Cis_thanked%2Cis_nothelp%3Bdata%5B%2A%5D.mark_infos%5B%2A%5D.url%3Bdata%5B%2A%5D.author.follower_count%2Cbadge%5B%3F%28type%3Dbest_answerer%29%5D.topics&limit={limit}&offset={offset}'
    headers = {
        'authorization': 'oauth c3cef7c66a1843f8b3a9e6a1e3160e20',
        'Host': 'www.zhihu.com',
        'Referer': 'https://www.zhihu.com/',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.87 Safari/537.36',
    }

    def start_requests(self):
        """
            重写爬虫入口 start_requests
        """
        url = 'https://www.zhihu.com/node/ExploreAnswerListV2?params='
        # offset = 0
        for offset in range(4):
            time.sleep(random.random())
            paramse = {"offset": offset * 5, "type": "day"}
            full_url = f'{url}{json.dumps(paramse)}'
            yield scrapy.Request(url=full_url, headers=self.headers, callback=self.parse_question)

    def parse_question(self, response):
        """
            处理 question
            这里只是提取了 question 部分字段信息，因为 question 页面是 JS 动态加载的，无法直接爬取
            有些字段须在 parse_answer 方法中提取
        """
        pattern = '(https://www.zhihu.com/question/\d+)/answer/\d+'
        question_links = response.xpath('//a[@class="question_link"]/@href').extract()
        question_links = [urljoin(response.url, url) for url in question_links]
        question_links = [re.findall(pattern, url)[0] for url in question_links]
        question_titles = response.xpath('//a[@class="question_link"]/text()').extract()

        for i, question_link in enumerate(question_links):
            question_id = re.findall('\d+', question_link)[0]
            item_loader = ItemLoader(item=ZhihuQuestionItem(), response=response)
            item_loader.add_value('zhihu_id', question_id)
            item_loader.add_value('url', question_link)
            item_loader.add_value('title', question_titles[i])
            question_item = item_loader.load_item()

            yield scrapy.Request(
                url=self.start_answer_url.format(question=question_id, limit=20, offset=0),
                headers=self.headers,
                callback=self.parse_answer
            )
            yield question_item

    def parse_answer(self, response):
        """
            处理 question 的 answer
        """
        data_json = json.loads(response.text)
        totals = data_json.get('paging').get('totals')
        is_end = data_json.get('paging').get('is_end')
        next_url = data_json.get('paging').get('next')

        # 添加 question 字段信息并 yield 到 pipeline
        # 通过 UPDATE 方式将字段信息更新到 zhihu_question 表中
        # question 字段信息并不完整，还有几个字段无法提取，因为需要在
        # js 加载的动态页面中提取，比较麻烦，这里只提取了静态页面中的数据
        question_item = ZhihuQuestionItem()
        question_item['zhihu_id'] = re.findall('questions/(\d+)/answers?', next_url)
        question_item['answer_num'] = totals
        question_item['create_time'] = data_json.get('data')[0].get('question').get('created')
        question_item['update_time'] = data_json.get('data')[0].get('question').get('updated_time')
        yield question_item

        # 提取 answer 信息
        for data in data_json.get('data'):
            answer_item = ZhihuAnswerItem()
            answer_item['zhihu_id'] = data.get('id')
            answer_item['url'] = data.get('url')
            answer_item['question_id'] = data.get('question').get('id')
            answer_item['author_id'] = data.get('author').get('id')
            answer_item['author_name'] = data.get('author').get('name')
            answer_item['content'] = data.get('content')
            answer_item['praise_num'] = data.get('voteup_count')
            answer_item['comments_num'] = data.get('comment_count')
            answer_item['create_time'] = data.get('created_time')
            answer_item['update_time'] = data.get('updated_time')
            # 将 answer 字段信息 yield 到 pipeline
            yield answer_item

        if not is_end:
            yield scrapy.Request(url=next_url, headers=self.headers, callback=self.parse_answer)

将爬取数据保存到 MySQL 中

通过 pipeline 管道将爬取的数据入库，有几种方案：

• ①通过一个 pipeline 来处理所有 item，上一章节中爬取了 jobbole 的文章，写过了一个 pipeline 来将数据保存到 MySQL，现在又爬去了 知乎的文章和回答信息，可以将新增的这两个 item 数据同样通过这个 pipeline 来处理，这里需要改造之前写好的 pipeline，已达到同时处理 3 个 不同的 item 的需求，并分别保存到相应的 MySQL 数据表当中
• ②每个 item 对应一个 pipeline，也就是一个 pipeline 只处理一个 item
• ③每个网站一个 pipeline，伯乐在线网站一个 pipeline，知乎一个 pipeline

分析三种不同方案利弊：

• 第二种和第三种方案差别不大，但是如果项目中爬虫很多，每个网站或者每个 item 都写一个 pipeline，而每个 pipeline 都去连接 MySQL 数据库进行操作，是很不合理的，过度占用资源
• 第一种方案虽然全部写在一个 pipeline 里面看上去好像逻辑没有那么清晰，但最合理，占用资源小，管理起来方便

还有一种情况就是一个项目中很多爬虫，有的爬虫数据需要存储到 MySQL，有的需要存储到 MongoDB，这种需要分别存储到不同数据库的情况，就可以采用每个数据库写一个 pipeline

• 思路1

在 pipeline 中，通过判断 item.__class__.__name__ 是否等于哪一个 item 类名（scrapy 中提供了 item.__class__.__name__ 来获取 item 所属类），来进行处理。
但是这样做把程序写的太死，item 类名就不能变了，如果变了，这里还要过来修改，会很麻烦

image.png

• 思路2

另外一种技巧，借鉴 Django 的 Model 思想

• Django 中对数据库的操作采用的是 Model 来完成的，每张表对应一个 class，而每个 class 都可以定义私有方法，这样就屏蔽掉具体的 SQL 语句
• 所以，可以将 pipeline 中不同爬虫变化的部分拿到 item 类定义中去，也就是把所有的 SQL 语句插入数据的操作拿到 item 类定义中去，定义成一个统一的方法 get_insert_sql，这样只需在 pipeline 中调用 这个函数就可以了，不同的 item 都调用各自的 get_insert_sql 方法，pipeline 中就不需要针对每个爬虫写对应处理逻辑，结构也更加清晰

pipeline 管道文件

# ArticleSpider/pipelines.py

# -*- coding: utf-8 -*-

# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://doc.scrapy.org/en/latest/topics/item-pipeline.html
import json

import pymysql
from twisted.enterprise import adbapi
from scrapy.pipelines.images import ImagesPipeline


class ArticlespiderPipeline(object):
    def process_item(self, item, spider):
        return item


class ArticleImagePipeline(ImagesPipeline):
    def item_completed(self, results, item, info):
        if 'front_img_url' in item:
            for ok, v in results:
                image_file_path = v['path']
                item['front_img_path'] = image_file_path
            return item


class JsonWithEncodingPipeline(object):
    def __init__(self):
        self.file = open('article.json', 'a', encoding='utf-8')

    def process_item(self, item, spider):
        self.file.write(json.dumps(dict(item), ensure_ascii=False) + '\n')
        return item

    def close_spider(self):
        self.file.close()


class MySQLPipeline(object):
    def __init__(self):
        self.conn = pymysql.connect('127.0.0.1', 'pythonic', 'pythonic', 'Articles', charset='utf8')
        self.cursor = self.conn.cursor()

    def process_item(self, item, spider):
        insert_sql = '''
                    insert into jobbole_article(title,create_date,url,url_object_id,front_image_url,
                        front_image_path,praise_nums,comment_nums,fav_nums,tags,content)
                    values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                '''
        self.cursor.execute(insert_sql, (
            item['title'], item['create_date'], item['url'], item['url_object_id'],
            item['front_img_url'], item['front_img_url'], item['praise_nums'],
            item['comment_nums'], item['fav_nums'], item['tags'], item['content']
        ))
        self.conn.commit()

    def close_spider(self):
        self.cursor.close()
        self.conn.close()


class MySQLTwistedPipeline(object):

    def __init__(self, dbpool):
        self.dbpool = dbpool

    # Scrapy 提供了一个类方法可以直接获取 settings.py 文件中的配置信息
    @classmethod
    def from_settings(cls, settings):
        db_params = dict(
            host=settings['MYSQL_HOST'],
            user=settings['MYSQL_USER'],
            password=settings['MYSQL_PASSWORD'],
            database=settings['MYSQL_DATABASE'],
            port=settings['MYSQL_PORT'],
            charset='utf8',
            cursorclass=pymysql.cursors.DictCursor,
        )

        # Twister 只是提供了一个异步的容器，并没有提供数据库连接，所以连接数据库还是要用 pymysql 进行连接
        # adbapi 可以将 MySQL 的操作变为异步
        # ConnectionPool 第一个参数是我们连接数据库所使用的 库名，这里是连接 MySQL 用的 pymysql
        # 第二个参数就是 pymysql 连接操作数据库所需的参数，这里将参数组装成字典 db_params，当作关键字参数传递进去
        dbpool = adbapi.ConnectionPool('pymysql', **db_params)
        return cls(dbpool)

    def process_item(self, item, spider):
        # 使用 Twisted 将 MYSQL 插入变成异步
        # 执行 runInteraction 方法的时候会返回一个 query 对象，专门用来处理异常
        query = self.dbpool.runInteraction(self.do_insert, item)
        # 添加错误处理方法到 query 对象
        # addErrback 第一个参数是处理异常的方法，后面的参数是这个方法所需的参数
        # 因为定义的 handle_error 方法需要接收 item、spider 参数，所以这里需要传递
        query.addErrback(self.handle_error, item, spider)

    def do_insert(self, cursor, item):
        # 执行具体的插入操作
        # 这里已经不需要手动 commit 了，Twisted 会自动 commit

        # 调用每个 item 的 get_insert_sql 方法，获取 insert_sql 语句
        # 以及 params 参数，完成数据的插入操作
        insert_sql, params = item.get_insert_sql()
        cursor.execute(insert_sql, params)

    def handle_error(self, failure, item, spider):
        # 异常处理方法，处理异步插入数据库时产生的异常
        # failure 参数不需要我们自己传递，出现异常会自动将异常当作这个参数传递进来
        # item、spider 参数并不是必须的，传递进来最大好处是如果出现异常
        # debug 的时候很容易通过 item 和 spider 来定位错误
        print(f'出现异常：{failure}')
        print(item.__class__.__name__)

修改 items.py 文件，将插入 MySQL 的语句封装成方法，放到每个 ITEM 类中

# ArticleSpider/items.py

# -*- coding: utf-8 -*-

# Define here the models for your scraped items
#
# See documentation in:
# https://doc.scrapy.org/en/latest/topics/items.html
import re
from datetime import datetime

import scrapy
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst, Join
from ArticleSpider.settings import SQL_DATETIME_FORMAT, SQL_DATE_FORMAT


# class ArticlespiderItem(scrapy.Item):
#     # define the fields for your item here like:
#     # name = scrapy.Field()
#     pass


# class JobBoleArticleItem(scrapy.Item):
#     title = scrapy.Field()  # 文章标题
#     create_date = scrapy.Field()  # 文章发布日期
#     url = scrapy.Field()  # 文章 URL
#     url_object_id = scrapy.Field()  # 文章 URL 的 MD5 值
#     front_img_url = scrapy.Field()  # 文章封面图(文章列表页显示的封面图，通常是文章详情第一张图片)
#     front_img_path = scrapy.Field()  # 文章封面图存放路径
#     praise_nums = scrapy.Field()  # 点赞数
#     comment_nums = scrapy.Field()  # 评论数
#     fav_nums = scrapy.Field()  # 收藏数
#     tags = scrapy.Field()  # 文章标签
#     content = scrapy.Field()  # 文章内容


# def add_jobbole(value):
#     return f'{value} --jobbole'


def convert_date(value):
    """
        将字符串转换成日期
    """
    try:
        return datetime.strptime(value, '%Y/%m/%d')
    except Exception as e:
        return datetime.now().date()


def take_nums(value):
    """
        提取数字
    """
    re_find = re.findall('.*?(\d+).*', value)
    if re_find:
        return int(re_find[0])
    else:
        return 0


def remove_tags_comment(value):
    """
        移除 tags 中的评论
    """
    return '' if '评论' in value else value


class ArticleItemLoader(ItemLoader):
    """
        自定义 ItemLoader，继承自 Scrapy 的 ItemLoader
        来改变 ItemLoader 的默认 output_processor 方法
    """
    default_output_processor = TakeFirst()


class JobBoleArticleLoadItem(scrapy.Item):
    title = scrapy.Field(
        # input_processor=MapCompose(add_jobbole, lambda x: x+'[article]'),
        # output_processor=TakeFirst()
    )
    create_date = scrapy.Field(
        input_processor=MapCompose(convert_date)
    )
    url = scrapy.Field()
    url_object_id = scrapy.Field()
    front_img_url = scrapy.Field(
        # 因为 使用 scrapy 自带的下载图片管道图片字段必须是可迭代的
        # 所以这里不再使用默认的 output_processor=TakeFirst()
        # 而是通过 lambda 表达式返回一个列表，因为本来传进来
        # 就是一个列表，所以什么也不需要做，只需要直接将值重新返回
        output_processor=MapCompose(lambda x: x)
    )
    front_img_path = scrapy.Field()
    praise_nums = scrapy.Field(
        input_processor=MapCompose(take_nums)
    )
    comment_nums = scrapy.Field(
        input_processor=MapCompose(take_nums)
    )
    fav_nums = scrapy.Field(
        input_processor=MapCompose(take_nums)
    )
    tags = scrapy.Field(
        input_processor=MapCompose(remove_tags_comment),
        output_processor=Join(',')
    )
    content = scrapy.Field()

    def get_insert_sql(self):
        # ON DUPLICATE KEY UPDATE 为 MySQL 独有的 '插入更新' 语法
        # 将一条数据插入 MySQL 时，如果这条数据已经存在于数据库中
        # 不会报错，而是执行更新操作，更新的字段可以自己指定
        insert_sql = '''
            INSERT INTO jobbole_article(title,create_date,url,url_object_id,front_image_url,
                front_image_path,praise_nums,comment_nums,fav_nums,tags,content)
            VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON DUPLICATE KEY UPDATE praise_nums=VALUES(praise_nums),comment_nums=VALUES(comment_nums),
                fav_nums=VALUES(fav_nums),content=VALUES(content),
        '''
        params = (
            self['title'], self['create_date'], self['url'], self['url_object_id'],
            self['front_img_url'], self['front_img_url'], self['praise_nums'],
            self['comment_nums'], self['fav_nums'], self['tags'], self['content']
        )
        return insert_sql, params


class ZhihuQuestionItem(scrapy.Item):
    """
        知乎问题 Item
    """
    # 问题 id
    zhihu_id = scrapy.Field()
    # 所属话题
    topics = scrapy.Field()
    # 问题 URL
    url = scrapy.Field()
    # 问题标题
    title = scrapy.Field(
        input_processor=MapCompose(lambda x: x.strip())
    )
    # 问题内容
    content = scrapy.Field()
    # 问题回答数量
    answer_num = scrapy.Field()
    # 问题评论数
    comments_num = scrapy.Field()
    # 问题关注数
    watch_user_num = scrapy.Field()
    # 问题点击数(浏览数)
    click_num = scrapy.Field()
    # 问题创建时间
    create_time = scrapy.Field()
    # 问题更新时间
    update_time = scrapy.Field()
    # 问题爬取时间
    crawl_time = scrapy.Field()

    def get_insert_sql(self):
        insert_sql = '''
            INSERT INTO zhihu_question(zhihu_id,topics,url,title,content,create_time,update_time,
                answer_num,comments_num,watch_user_num,click_num,crawl_time,crawl_update_time)
            VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON DUPLICATE KEY UPDATE create_time=VALUES(create_time),update_time=VALUES(update_time),
                answer_num=VALUES(answer_num),crawl_update_time=VALUES(crawl_update_time)
        '''

        # 因为 zhihu_question 爬虫中使用的是 scrapy 的 ItemLoader
        # 而没有使用自定义的 ArticleItemLoader，并且没有在上面定义
        # 字段里面使用 output_processor = TakeFirst()，所以现在每个
        # 字段的类型都是列表，所以其实也是可以在这个地方插入数据库
        # 之前将数据字段类型全部修改成对应于 MySQL 表中的字段类型
        zhihu_id = int(self['zhihu_id'][0])
        topics = None
        url = self['url'][0] if self.get('url') else ''
        title = self['title'][0] if self.get('title') else ''
        content = None
        create_time = datetime.fromtimestamp(self['create_time']).strftime(SQL_DATETIME_FORMAT) if self.get('create_time') else None
        update_time = datetime.fromtimestamp(self['update_time']).strftime(SQL_DATETIME_FORMAT) if self.get('update_time') else None
        answer_num = self['answer_num'] if self.get('answer_num') else 0
        comments_num = 0
        watch_user_num = 0
        click_num = 0
        # 因为 执行 insert_sql 插入数据时，需要传递字符串类型，所以将时间转换成字符串
        crawl_time = datetime.now().strftime(SQL_DATETIME_FORMAT)
        crawl_update_time = datetime.now().strftime(SQL_DATETIME_FORMAT)

        params = (
            zhihu_id, topics, url, title, content, create_time, update_time, answer_num,
            comments_num, watch_user_num, click_num, crawl_time, crawl_update_time
        )
        return insert_sql, params


class ZhihuAnswerItem(scrapy.Item):
    """
        知乎问题回答 Item
    """
    # 回答 id
    zhihu_id = scrapy.Field()
    # 回答 URL
    url = scrapy.Field()
    # 所属问题 id
    question_id = scrapy.Field()
    # 用户 id
    author_id = scrapy.Field()
    # 用户名
    author_name = scrapy.Field()
    # 回答内容
    content = scrapy.Field()
    # 赞同数
    praise_num = scrapy.Field()
    # 评论数
    comments_num = scrapy.Field()
    # 回答创建时间
    create_time = scrapy.Field()
    # 回答修改时间
    update_time = scrapy.Field()
    # 回答爬取时间
    crawl_time = scrapy.Field()

    def get_insert_sql(self):
        insert_sql = '''
            INSERT INTO zhihu_answer(zhihu_id,url,question_id,author_id,author_name,
                content,praise_num,comments_num,create_time,update_time,crawl_time,crawl_updatetime)
            VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON DUPLICATE KEY UPDATE content=VALUES(content),praise_num=VALUES(praise_num),
                comments_num=VALUES(comments_num),update_time=VALUES(update_time),
                crawl_updatetime=VALUES(crawl_updatetime)
        '''

        # create_time 和 update_time 从 spider 传递过来是 int 类型
        # 先通过 datetime.fromtimestamp() 将 int 转换成 datetime 类型
        # 再通过 strftime() 将 datetime 类型转成 str 类型
        create_time = datetime.fromtimestamp(self['create_time']).strftime(SQL_DATETIME_FORMAT)
        update_time = datetime.fromtimestamp(self['update_time']).strftime(SQL_DATETIME_FORMAT)
        crawl_time = datetime.now().strftime(SQL_DATETIME_FORMAT)
        crawl_updatetime = datetime.now().strftime(SQL_DATETIME_FORMAT)

        params = (
            self['zhihu_id'], self['url'], self['question_id'], self['author_id'],
            self['author_name'], self['content'], self['praise_num'], self['comments_num'],
            create_time, update_time, crawl_time, crawl_updatetime
        )
        return insert_sql, params