1.3 Pei Zhi Shi Kai Fa Template

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

1. 定义模型

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

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

2. 添加配置

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

init: 关于初始化的配置
download: 关于下载请求的配置
parse: 解析配置
pipeline: 存储配置
events: 事件配置

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

2.1 配置初始化: `init`

inti 配置为一个列表, 列表中传入 template.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}

import time
 
from bricks.db.redis_ import Redis
from bricks.plugins.make_seeds import by_redis
from bricks.spider import template
from bricks.spider.template import Config
 
# 需自行配置 redis
from no_views.conn import redis_config
 
 
class Spider(template.Spider):
 
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 需自行配置 redis
        self.redis = Redis(
            **redis_config
        )
 
    @property
    def config(self) -> Config:
        return Config(
            init=[
                template.Init(
                    func=by_redis,
                    kwargs={
                        'conn': self.redis,
                        'path': '*|*',
                        'key_type': 'string',
                    },
                    layout=template.Layout(
                        factory={
                            "time": lambda: time.time()
                        }
                    )
                )
            ]
        )

2.2 配置下载: `download`

与 init 类似, 配置为一个列表, 需传入 template.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: 最大重试次数

import time
 
from bricks.db.redis_ import Redis
from bricks.plugins.make_seeds import by_redis
from bricks.spider import template
from bricks.spider.template import Config
from no_views.conn import redis_config
 
 
class Spider(template.Spider):
 
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.redis = Redis(
            **redis_config
        )
 
    @property
    def config(self) -> Config:
        return Config(
            init=[
                template.Init(
                    func=by_redis,
                    kwargs={
                        # 需自行配置 redis
                        'conn': self.redis,
                        'path': '*|*',
                        'key_type': 'string',
                    },
                    layout=template.Layout(
                        factory={
                            "time": lambda: time.time()
                        }
                    )
                )
            ],
            download=[
                template.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.3 配置解析: `parse`

与以上类似, 配置 template.Parse, 参数配置为:

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

import time
 
from bricks.db.redis_ import Redis
from bricks.plugins.make_seeds import by_redis
from bricks.spider import template
from bricks.spider.template import Config
from no_views.conn import redis_config
 
 
class Spider(template.Spider):
 
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.redis = Redis(
            **redis_config
        )
 
    @property
    def config(self) -> Config:
        return Config(
            init=[
                template.Init(
                    func=by_redis,
                    kwargs={
                        # 需自行配置 redis
                        'conn': self.redis,
                        'path': '*|*',
                        'key_type': 'string',
                    },
                    layout=template.Layout(
                        factory={
                            "time": lambda: time.time()
                        }
                    )
                )
            ],
            download=[
                template.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=[
                template.Parse(
                    func="json",
                    kwargs={
                        "rules": {
                            "data.list": {
                                "userId": "userId",
                                "roomId": "roomId",
                                "score": "score",
                                "startTime": "startTime",
                                "kugouId": "kugouId",
                                "status": "status",
                            }
                        }
                    },
                    layout=template.Layout(
                        rename={"userId": "user_id"},
                        default={"modify_at": time.time(), "page": "{page}", "seeds_time": "{time}"}
                    )
                )
            ]
        )

2.4 配置存储: `pipeline`

配置 template.Pipeline, 参数为:

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

import time
 
from bricks.db.redis_ import Redis
from bricks.plugins.make_seeds import by_redis
from bricks.spider import template
from bricks.spider.template import Config
from no_views.conn import redis_config
 
 
class Spider(template.Spider):
 
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.redis = Redis(
            **redis_config
        )
 
    @property
    def config(self) -> Config:
        return Config(
            init=[
                template.Init(
                    func=by_redis,
                    kwargs={
                        # 需自行配置 redis
                        'conn': self.redis,
                        'path': '*|*',
                        'key_type': 'string',
                    },
                    layout=template.Layout(
                        factory={
                            "time": lambda: time.time()
                        }
                    )
                )
            ],
            download=[
                template.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=[
                template.Parse(
                    func="json",
                    kwargs={
                        "rules": {
                            "data.list": {
                                "userId": "userId",
                                "roomId": "roomId",
                                "score": "score",
                                "startTime": "startTime",
                                "kugouId": "kugouId",
                                "status": "status",
                            }
                        }
                    },
                    layout=template.Layout(
                        rename={"userId": "user_id"},
                        default={"modify_at": time.time(), "page": "{page}", "seeds_time": "{time}"}
                    )
                )
            ],
            pipeline=[
                template.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: 存储之后执行的事件

具体的配置样例为:

events={
    const.AFTER_REQUEST: [
        template.Task(
            func=scripts.is_success,
            kwargs={
                "match": [
                    "context.response.get('code') == 0"
                ]
            }
        ),
    ],
    const.BEFORE_PIPELINE: [
        template.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.db.redis_ import Redis
from bricks.lib.queues import RedisQueue
from bricks.plugins import scripts
from bricks.plugins.make_seeds import by_redis
from bricks.spider import template
from bricks.spider.template import Config
from no_views.conn import redis_config
 
 
class Spider(template.Spider):
 
  def __init__(self, **kwargs):
    super().__init__(**kwargs)
    self.redis = Redis(
      **redis_config
    )
 
  @property
  def config(self) -> Config:
    return Config(
      init=[
        template.Init(
          func=by_redis,
          kwargs={
            # 需自行配置 redis
            'conn': self.redis,
            'path': '*|*',
            'key_type': 'string',
          },
          layout=template.Layout(
            factory={
              "time": lambda: time.time()
            }
          )
        )
      ],
      download=[
        template.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=[
        template.Parse(
          func="json",
          kwargs={
            "rules": {
              "data.list": {
                "userId": "userId",
                "roomId": "roomId",
                "score": "score",
                "startTime": "startTime",
                "kugouId": "kugouId",
                "status": "status",
              }
            }
          },
          layout=template.Layout(
            rename={"userId": "user_id"},
            default={"modify_at": time.time(), "page": "{page}", "seeds_time": "{time}"}
          )
        )
      ],
      pipeline=[
        template.Pipeline(
          func=lambda context: print(context.items),
          success=True
        )
      ],
      events={
        const.AFTER_REQUEST: [
          template.Task(
            func=scripts.is_success,
            kwargs={
              "match": [
                "context.response.get('code') == 0"
              ]
            }
          )
        ],
        const.BEFORE_PIPELINE: [
          template.Task(
            func=scripts.turn_page,
            kwargs={
              "match": [
                "context.response.get('data.hasNextPage') == 1"
              ]
            }
          )
        ]
      }
    )
 
 
if __name__ == '__main__':
  spider = Spider(
    task_queue=RedisQueue(**redis_config),
  )
  spider.run(
    task_name='all',
  )

1.2 代码式开发 1.4 配置式开发(form)

1. 定义模型

2. 添加配置

2.1 配置初始化: init

2.2 配置下载: download

2.3 配置解析: parse

2.4 配置存储: pipeline

2.5 配置事件: events