1.4 Pei Zhi Shi Kai Fa Form

本文以前文中的需求以及代码式开发为例, 如果你还不了解, 可以跳转查看

1. 定义模型

首先, 你需要导入 bricks 爬虫基类, 然后定义一个类继承这个基类, 本文以配置式开发(form)为例, 因此我们需要导入的基类的地址为: bricks.spider.form.Spider

from bricks.spider.form import Spider
 
 
class MySpider(Spider):
    pass

2. 添加配置

Spider 的配置是 bricks.spider.form.Config, 有以下几个属性

init: 关于初始化的配置
spider: 爬虫流程控制
events: 事件配置

from bricks.spider.form import Spider, Config
 
 
class MySpider(Spider):
    @property
    def config(self) -> Config:
        return Config()

2.1 配置初始化: `init`

inti 配置为一个列表, 列表中传入 form.Init 类型, 其配置参数作用为:

func: 初始化函数, 在初始化时会调用该函数, 测试时可以填写一个 lambda 表达式, bricks 目前封装了一些内置方法, 在 bricks.plugins.make_seeds 中, 如 by_csv, by_mongo, by_redis, by_sqlite
args: 初始化函数的位置参数
kwargs: 初始化函数的关键词参数
layout: 一些修改参数
- rename: 将结果中的某些键值删除替换为另一键值, 例: {"a": "b"}, 将结果中的 a 键值改为 b 键值
- default: 为结果中的某个键设置默认值, 例: {"a": 1000}
- factory: 将结果中的某个值作为参数传入工厂函数中, 例: {"a": lambda x: x + 1}
- show: 是否展示结果中的某个键值, 例: {"a": True, "b": lambda: x: x> 10}

from bricks.spider.form import Spider, Config, Init
 
 
class MySpider(Spider):
    @property
    def config(self) -> Config:
        return Config(
 
            init=[
                Init(
                    func=lambda: {"page": 1}
                )
            ]
 
        )

2.2 配置爬虫流程: `spider`

与 init 类似, 配置为一个列表, 列表内可用的节点类型有: form.Download|form.Parse|form.Pipeline|form.Task, 顺序可以自由组合

因为 template 是流程固化的配置式爬虫, 而 form 是流程自由的配置式爬虫, 你可以在 form 中自由的配置流程, 也可以在 form 中使用 template 的配置例如, 我可以在 spider 配置内配置如下流程: 下载配置 1 -> 解析配置 1 -> 其他任务节点 1 -> 下载配置 2 -> 解析配置 2 -> 存储配置 1 -> 其他任务节点 2, 也就是说, 流程是可以自定义的. 你甚至可以使用 Task 节点代替 events 配置, 可以清晰的看到爬虫运行的每一个流程!

2.2.1 配置下载: `download`

配置节点为: form.Download, 参数配置为:

url: 请求地址
params: 请求参数
method: 请求方法
body: 请求体, 所有的 body, data, json 都在这里填入, 会根据 headers 中的 Content-Type 自动判断
headers: 请求头
cookies: 请求 cookies
options: 其他配置, 包括 auth 等等, 下载器需要的额外参数也放在这里
timeout: 超时时间
allow_redirects: 是否允许重定向
proxies: 代理, 形式为: "http://ip:prot"
proxy: 代理键, 代理键会在 bricks 中自动获取代理, 形式为: {"ref": ...(代理类), ...(连接参数)}
is_success: 通过状态码判断请求是否通过, 形式为脚本字符串: "response.status_code in [200, 302]"
retry: 重试次数起始值
max_retry: 最大重试次数
archive: 是否进行存档, 如果进行存档, 则重试的时候不会再执行当前节点前面的所有流程, 就类似闯关模式, 这关过去了存档后不会继续闯这一关, 而是从下一关开始闯

from bricks.spider.form import Spider, Config, Init, Download
 
 
class MySpider(Spider):
    @property
    def config(self) -> Config:
        return Config(
 
            init=[
                Init(
                    func=lambda: {"page": 1}
                )
            ],
            spider=[
                Download(
                    url="https://fx1.service.kugou.com/mfanxing-home/h5/cdn/room/index/list_v2",
                    params={
                        "page": "{page}",
                        "cid": 6000
                    },
                    headers={
                        "User-Agent": "@chrome",
                        "Content-Type": "application/json;charset=UTF-8",
                    }
                )
            ]
 
        )

2.2.2 配置解析: `parse`

配置节点为 form.Parse, 参数配置为:

func: 解析函数, 在解析时会调用该函数, 或者填入框架默认支持的解析器, 如: json, xpath, jsonpath, regex
args: 解析函数的位置参数
kwargs: 解析函数的关键词参数
layout: 一些修改参数, 与 init 一致

import time
 
from bricks.spider.form import Spider, Config, Init, Download, Parse, Layout
 
 
class MySpider(Spider):
    @property
    def config(self) -> Config:
        return Config(
 
            init=[
                Init(
                    func=lambda: {"page": 1}
                )
            ],
            spider=[
                Download(
                    url="https://fx1.service.kugou.com/mfanxing-home/h5/cdn/room/index/list_v2",
                    params={
                        "page": "{page}",
                        "cid": 6000
                    },
                    headers={
                        "User-Agent": "@chrome",
                        "Content-Type": "application/json;charset=UTF-8",
                    }
                ),
                Parse(
                    func="json",
                    kwargs={
                        "rules": {
                            "data.list": {
                                "userId": "userId",
                                "roomId": "roomId",
                                "score": "score",
                                "startTime": "startTime",
                                "kugouId": "kugouId",
                                "status": "status",
                            }
                        }
                    },
                    layout=Layout(
                        rename={"userId": "user_id"},
                        default={"modify_at": time.time(), "page": "{page}", "seeds_time": "{time}"}
                    )
                )
            ]
 
        )

2.2.3 配置存储: `pipeline`

配置节点为 form.Pipeline, 参数为:

func: 存储函数, 在解析时会调用该函数, bricks 目前封装了一些内置方法, 在 bricks.plugins.storage 中, 如 to_csv, to_mongo, to_redis, to_sqlite
args: 存储函数的位置参数
kwargs: 存储函数的关键词参数
layout: 一些修改参数, 与 init 一致
success: 是否在存储成功后将对应种子删除

import time
 
from bricks.spider.form import Spider, Config, Init, Download, Parse, Layout, Pipeline
 
 
class MySpider(Spider):
    @property
    def config(self) -> Config:
        return Config(
 
            init=[
                Init(
                    func=lambda: {"page": 1}
                )
            ],
            spider=[
                Download(
                    url="https://fx1.service.kugou.com/mfanxing-home/h5/cdn/room/index/list_v2",
                    params={
                        "page": "{page}",
                        "cid": 6000
                    },
                    headers={
                        "User-Agent": "@chrome",
                        "Content-Type": "application/json;charset=UTF-8",
                    }
                ),
                Parse(
                    func="json",
                    kwargs={
                        "rules": {
                            "data.list": {
                                "userId": "userId",
                                "roomId": "roomId",
                                "score": "score",
                                "startTime": "startTime",
                                "kugouId": "kugouId",
                                "status": "status",
                            }
                        }
                    },
                    layout=Layout(
                        rename={"userId": "user_id"},
                        default={"modify_at": time.time(), "page": "{page}", "seeds_time": "{time}"}
                    )
                ),
                Pipeline(
                    func=lambda context: print(context.items),
                    success=True
                )
            ]
 
        )

2.5 配置事件: `events`

events 为一个字典套列表套 task , 其中 events 的键为 bricks 内部的事件类型, 引用地址为: bricks.state.const, 一般常用的事件为:

BEFORE_REQUEST: 下载之前执行的事件
AFTER_REQUEST: 下载之后执行的事件
BERORE_PIPELINE: 存储之前执行的事件
AFTER_PIPELINE: 存储之后执行的事件

具体的配置样例为:

from bricks import const
from bricks.plugins import scripts
from bricks.spider.form import Task
 
events = {
    const.AFTER_REQUEST: [
        Task(
            func=scripts.is_success,
            kwargs={
                "match": [
                    "context.response.get('code') == 0"
                ]
            }
        ),
    ],
    const.BEFORE_PIPELINE: [
        Task(
            func=scripts.turn_page,
            kwargs={
                "match": [
                    "context.response.get('data.hasNextPage') == 1"
                ],
            }
        ),
    ]
}

如以上代码, 则会在请求之后执行 scripts.is_success 方法, 在存储之前执行 scripts.turn_page 方法, 这些方法默认会接受一个 context 参数(由 bricks 自动生成传递), 该参数为请求的上下文, 包含当前事件为止的所有属性, 如 request, response 等...

3 整体代码

import time
 
from bricks import const
from bricks.plugins import scripts
from bricks.spider.form import Spider, Config, Init, Download, Parse, Layout, Pipeline, Task
 
 
class MySpider(Spider):
    @property
    def config(self) -> Config:
        return Config(
 
            init=[
                Init(
                    func=lambda: {"page": 1}
                )
            ],
            spider=[
                Download(
                    url="https://fx1.service.kugou.com/mfanxing-home/h5/cdn/room/index/list_v2",
                    params={
                        "page": "{page}",
                        "cid": 6000
                    },
                    headers={
                        "User-Agent": "@chrome",
                        "Content-Type": "application/json;charset=UTF-8",
                    }
                ),
                Parse(
                    func="json",
                    kwargs={
                        "rules": {
                            "data.list": {
                                "userId": "userId",
                                "roomId": "roomId",
                                "score": "score",
                                "startTime": "startTime",
                                "kugouId": "kugouId",
                                "status": "status",
                            }
                        }
                    },
                    layout=Layout(
                        rename={"userId": "user_id"},
                        default={"modify_at": time.time(), "page": "{page}", "seeds_time": "{time}"}
                    )
                ),
                Pipeline(
                    func=lambda context: print(context.items),
                    success=True
                )
            ],
            events={
                const.AFTER_REQUEST: [
                    Task(
                        func=scripts.is_success,
                        kwargs={
                            "match": [
                                "context.response.get('code') == 0"
                            ]
                        }
                    ),
                ],
                const.BEFORE_PIPELINE: [
                    Task(
                        func=scripts.turn_page,
                        kwargs={
                            "match": [
                                "context.response.get('data.hasNextPage') == 1"
                            ],
                        }
                    ),
                ]
            }
        )
 
 
if __name__ == '__main__':
    spider = MySpider()
    spider.run()

1.3 配置式开发(template)2.1.1 概念初拾

1. 定义模型

2. 添加配置

2.1 配置初始化: init

2.2 配置爬虫流程: spider

2.2.1 配置下载: download

2.2.2 配置解析: parse

2.2.3 配置存储: pipeline

2.5 配置事件: events