简书爬虫项目

爬取内容：

　　简书中每一篇文章的具体数据，主要包括文章标题、钻石数、发表日期、文章字数、文章阅读量、文章的评论量和点赞量等，这里爬取2000条左右保存至数据库

代码实现在文末！

分析思路：

• 首先，谷歌浏览器抓包，获取简书首页加载新文章的方式，当我们点击主页的加载更多的按钮的时候，后台发送了一个异步的POST请求来加载新文章

请求有一个page参数，即是我们需要的。而且可以观察到返回的并非json数据，而是html数据

 接下来，我们做的就是从返回的html中找出跳到具体文章内容的链接，这里过程就不加赘述，一会会在代码里看到

• 所以我们可有有如下的一个逻辑过程：
• 通过向简书发送POST请求，获取新的文章列表的html，当然参数page要变，以控制页码
• 从返回的html代码中解析出真正文章的url
• 向真正的文章页面发送请求，获取响应
• 从响应的页面代码中提取出我们想要的文章数据
• 整理文章数据后保存至数据库

相关库与方法：

• Python网络请求requests库
• 解析方法xpath，这里用到的是Python的lxml库
• 数据库链接mysql的库，pymysql
• 多线程及队列，Python的threading和queue模块

代码结构及实现：

# 简书爬虫项目

### 目录结构
- lilei_spider
    - ____init____.py
    - jianshu_index.py
    - config.py
    - storage.py
    - utils
        - db.py
        - tools.py
        
## 文件说明：
#### jianshu_index.py
> 简书爬虫整体逻辑实现，也是运行项目的入口文件

#### config.py
> 项目中用到的配置参数

#### storage.py
> 实现数据的存储

#### db.py
> 数据库连接相关

#### tools.py
> 工具函数及工具类

#### article.sql
> 导出的数据库文件

#### requirements.txt
> 项目依赖库

### 技术说明
- python网络请求库：requests
- 数据解析方法：xpath(lxml库)
- 连接数据库：pymysql
- 多线程：threading，queue

代码实现：

jianshu_index.py

import requests
import threading
from queue import Queue
from lxml import etree

from config import INDEX_ADDRESS as index_address
from config import ADDRESS as address
from config import HEADERS as headers
from storage import Storage
from utils.tools import DecodingUtil


class JianshuSpider(object):
    """爬取简书首页数据的爬虫类"""
    def __init__(self):
        self.max_page = 300  # 爬取总文章数：300*7=2100
        self.params_queue = Queue()  # 存放地址发送post请求参数的队列
        self.url_queue = Queue()  # 存放文章url的队列
        self.index_queue = Queue()  # 存放首页响应的新文章列表的队列
        self.article_queue = Queue()  # 存放文章响应内容的队列
        self.content_queue = Queue()  # 存放格式化后的文章数据内容的队列

    def get_params_list(self):
        """构造post请求的page参数队列"""
        for i in range(1, self.max_page+1):
            self.params_queue.put({'page': i})

    def pass_post(self):
        """发送POST请求，获取新的文章列表，请求参数从队列中取出"""
        while True:
            response = requests.post(index_address, data=self.params_queue.get(), headers=headers)
            self.index_queue.put(DecodingUtil.decode(response.content))  # 返回结果放入队列
            self.params_queue.task_done()  # 计数减一
            print('pass_post', '@'*10)

    def parse_url(self):
        """根据首页返回的新的文章列表，解析出文章对应的url"""
        while True:
            content = self.index_queue.get()  # 从队列中取出一次POST请求的文章列表数据
            html = etree.HTML(content)
            a_list = html.xpath('//a[@class="title"]/@href')  # 每个li标签包裹着一篇文章
            for a in a_list:
                url = a  # xpath解析出文章的相对路径
                article_url = address + url
                self.url_queue.put(article_url)  # 放入队列
            self.index_queue.task_done()
            print('parse_url', '@'*10)

    def pass_get(self):
        """发送GET请求，获取文章内容页"""
        while True:
            article_url = self.url_queue.get()  # 从队列中获取文章的url
            response = requests.get(article_url, headers=headers)
            self.article_queue.put(DecodingUtil.decode(response.content))  # 返回结果放入队列
            self.url_queue.task_done()
            print('pass_get', '@'*10)

    def get_content(self):
        while True:
            article = dict()
            article_content = self.article_queue.get()
            html = etree.HTML(article_content)
            # 标题：title，钻石：diamond，创建时间：create_time，字数：word_number
            # 阅读量：read_number，评论数：comment_number，点赞数：like_number，文章内容：content
            article['title'] = html.xpath('//h1[@class="_2zeTMs"]/text()')[0].strip('
').strip('	')
            try:
                article['diamond'] = html.xpath('//span[@class="_3tCVn5"]/span/text()')[0]
            except IndexError:
                article['diamond'] = ''
            article['create_time'] = html.xpath('//div[@class="s-dsoj"]/time/text()')[0].replace('
', '')
            article['word_number'] = html.xpath('//div[@class="s-dsoj"]/span[2]/text()')[0].split(' ')[-1]
            article['read_number'] = html.xpath('//div[@class="s-dsoj"]/span[last()]/text()')[0].split(' ')[-1]
            article['comment_number'] = html.xpath('//div[@class="_3nj4GN"][1]/span/text()[last()]')[0]
            article['like_number'] = html.xpath('//div[@class="_3nj4GN"][last()]/span/text()[last()]')[0]
            content = html.xpath('//article[@class="_2rhmJa"]')  # html富文本内容
            article['content'] = DecodingUtil.decode(etree.tostring(content[0], method='html', encoding='utf-8'))
            self.content_queue.put(article)  # 放入队列
            self.article_queue.task_done()  # 上一队列计数减一
            print('get_content', '@'*10)

    def save(self):
        """保存数据"""
        while True:
            article_info = self.content_queue.get()  # 队列中获取文章信息
            # print(article_info)
            Storage.save_to_mysql(article_info)  # 文章数据保存到mysql数据库
            self.content_queue.task_done()
            print('save', '*'*20)

    def run(self):
        # 0.各个方法之间利用队列来传送数据
        # 1.简书首页加载新数据方式为POST请求，url不变，参数page变化，所以首先构造一个params集
        # 2.遍历params集发送POST请求，获取响应
        # 3.根据每一次获取的文章列表，再获取对应的真正文章内容的页面url
        # 4.向文章内容页面发送请求，获取响应
        # 5.提取对应的数据
        # 6.保存数据，一份存入数据库，一份存入excel
        thread_list = list()  # 模拟线程池
        t_params = threading.Thread(target=self.get_params_list)
        thread_list.append(t_params)
        for i in range(2):  # 为post请求开启3个线程
            t_pass_post = threading.Thread(target=self.pass_post)
            thread_list.append(t_pass_post)
        for j in range(2):  # 为解析url开启3个线程
            t_parse_url = threading.Thread(target=self.parse_url)
            thread_list.append(t_parse_url)
        for k in range(5):  # 为get请求开启5个线程
            t_pass_get = threading.Thread(target=self.pass_get)
            thread_list.append(t_pass_get)
        for m in range(5):  # 为提取数据开启5个线程
            t_get_content = threading.Thread(target=self.get_content)
            thread_list.append(t_get_content)
        # for n in range(5):  # 为保存数据开启5个线程
        t_save = threading.Thread(target=self.save)  # 保存数据一个线程
        thread_list.append(t_save)
        # =====================================================================================================
        for t in thread_list:
            t.setDaemon(True)  # 把子线程设置为守护线程，主线程结束，子线程结束
            t.start()
        for q in [self.params_queue, self.url_queue, self.index_queue, self.article_queue, self.content_queue]:
            q.join()  # 让主线程等待阻塞，等待队列的任务完成之后再结束
        print('主线程结束......')


if __name__ == '__main__':
    jianshu_spider = JianshuSpider()
    jianshu_spider.run()

config.py

import os


ADDRESS = 'https://www.jianshu.com'  # 简书网地址

INDEX_ADDRESS = 'https://www.jianshu.com/trending_notes'  # 首页加载数据的地址

HEADERS = {
    'x-pjax': 'true',
    'referer': 'https://www.jianshu.com/',
    'Content-Type': 'application/x-www-form-urlencoded',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                  'Chrome/77.0.3865.120 Safari/537.36',
    'x-csrf-token': 'PRFNi9/FDZmm/bV4f8ZueVNFln0PpQ5kgsMcSERpwpNugy/bcOBgNEZvBo4/aTwrm28awdmuTfcMaHcogJ1mdA=='
}

storage.py 

import pandas as pd
from utils.db import conn
from utils.tools import rep_invalid_char


class Storage(object):
    """存储数据"""
    cursor = conn.cursor()
    excel_writer = pd.ExcelWriter('article.xlsx')

    def __init__(self):
        pass

    @classmethod
    def save_to_mysql(cls, article:dict):
        """字典类型的文章数据保存到数据库"""
        title = article.get('title', '')
        diamond = article.get('diamond', '')
        create_time = article.get('create_time', '')
        word_number = article.get('word_number', '')
        read_number = article.get('read_number', '')
        comment_number = article.get('comment_number', '')
        like_number = article.get('like_number', '')
        content = rep_invalid_char(article.get('content', ''))
        # content = article.get('content', '')

        sql = "INSERT INTO `article2` (`title`,`diamond`,`create_time`,`word_number`,`read_number`,`comment_number`," 
              "`like_number`,`content`) VALUES ('{}','{}','{}','{}','{}','{}','{}','{}');".format(title, diamond,
              create_time, word_number, read_number, comment_number, like_number, content)
        try:
            cls.cursor.execute(sql)
            conn.commit()
        except Exception as e:
            print(e)
            # raise RuntimeError('保存至数据库过程Error！')
            print('保存至数据库过程Error！')

db.py

import pymysql


class DB(object):
    """数据库连接"""
    _host = '127.0.0.1'
    _user = 'root'
    _password = '******'
    _db = 'homework'

    @classmethod
    def conn_mysql(cls):
        return pymysql.connect(host=cls._host, user=cls._user, password=cls._password, db=cls._db, charset='utf8')


conn = DB.conn_mysql()

tools.py　　　

import re
import chardet


class DecodingUtil(object):
    """解码工具类"""
    @staticmethod
    def decode(content):
        """
        读取字节数据判断其编码，并正确解码
        :param content: 传入的字节数据
        :return: 正确解码后的数据
        """
        # print(chardet.detect(content))
        the_encoding = chardet.detect(content)['encoding']
        try:
            return content.decode(the_encoding)
        except UnicodeDecodeError:
            print('解码Error!')
            try:
                return content.decode('utf-8')
            except:
                return '未能成功解码的文章内容！'


def rep_invalid_char(old:str):
    """mysql插入操作时，有无效字符，替换"""
    invalid_char_re = r"[/?\[]*:]"
    return re.sub(invalid_char_re, "_", old)