跳转至

API Reference

本页由 mkdocstrings 从源码 docstring 自动生成。

maize.Request

Python
Request(
    url: str,
    *,
    method: Method | None = Method.GET,
    callback: Callable | None = None,
    error_callback: Callable | None = None,
    priority: int = 0,
    headers: dict | None = None,
    headers_func: Callable[[], Awaitable[dict]]
    | None = None,
    params: dict | None = None,
    data: dict | str | None = None,
    json: dict | None = None,
    cookies: dict | list[dict[str, Any]] | None = None,
    proxy: str | None = None,
    proxy_username: str | None = None,
    proxy_password: str | None = None,
    encoding: str | None = "utf-8",
    meta: dict | None = None,
    follow_redirects: bool = True,
    max_redirects: int = 20,
)

请求

:param url: 待抓取的url :param method: 请求方式，如 Method.GET, Method.POST, Method.PUT，默认 Method.GET :param callback: 自定义的解析函数，默认为 parse :param error_callback: 自定义的错误回调函数 :param priority: 请求优先级，默认为 0 :param headers: 请求头 :param params: 请求参数 :param data: 请求 body :param json: dict 类型的参数 :param cookies: 请求 cookies :param proxy: 代理ip :param proxy_username: 代理 ip 用户名 :param proxy_password: 代理 ip 密码 :param encoding: 编码，默认utf-8，当无法解析时，使用响应中的编码 :param meta: 自定义数据 :param follow_redirects: 是否允许重定向 :param max_redirects: 最大重定向次数，默认为 20

maize.Response

Python
Response(
    url: str,
    *,
    headers: dict[str, Any],
    request: Request,
    body: bytes = b"",
    text: str = "",
    status: int = 200,
    cookie_list: list[dict[str, Union[str, bool]]]
    | None = None,
    driver: Driver | None = None,
    source_response: R | None = None,
)

Bases: Generic[Driver, R]

响应

:param url: url :param headers: 响应头 :param request: 请求 Request :param body: 响应体 bytes 类型 :param text: 响应体 str 类型 :param status: 响应状态码，如 200 :param cookie_list: cookie 列表 :param driver: :param source_response: 原始响应，如下载器是 httpx，则为 httpx.Response 类型的实例。rpa 爬虫时，该字段为 None

maize.Item

Bases: BaseModel

maize.Field module-attribute

Python
Field = PydanticField

maize.Spider

Python
Spider()

Bases: StandardSpiderInterface

close async

Python
close()

在 Spider 关闭时执行一些清理操作

:return:

is_pause

Python
is_pause()

爬虫是否暂停

:return: 暂停 True，否则 False

open async

Python
open()

在 Spider 启动时执行一些初始化操作

:return:

parse async

Python
parse(response: Response)

自定义解析方法

:param response: 封装后的响应实例 :return:

pause_spider async

Python
pause_spider(lte_priority: int | None = None)

暂停爬虫

:param lte_priority: 请求优先级，默认为 None。为 None 时暂停采集所有的请求，否则只暂停采集小于 priority 的请求 :return:

proceed_spider async

Python
proceed_spider(gte_priority: int | None = None)

继续爬虫

:param gte_priority: 请求优先级，默认为 None。为 None 时继续采集所有的请求，否则只继续采集大于等于 priority 的请求 :return:

maize.TaskSpider

Python
TaskSpider()

Bases: Spider

start_requests abstractmethod async

Python
start_requests() -> typing.AsyncGenerator[
    Request, typing.Any
]

生成任务请求 :return:

maize.SpiderSettings

Bases: BaseSettings

爬虫配置主类

redis_url property

Python
redis_url: str

生成 Redis 连接 URL（向后兼容）

merge_settings

Python
merge_settings(settings: SpiderSettings) -> SpiderSettings

合并另一个 Settings 实例的配置到当前实例

:param settings: 要合并的 Settings 实例 :return: 当前实例

merge_settings_from_dict

Python
merge_settings_from_dict(
    settings_dict: dict[str, Any],
) -> SpiderSettings

合并 dict 的配置到当前实例

:param settings_dict: dict 类型的配置 :return: 当前实例

settings_customise_sources classmethod

Python
settings_customise_sources(
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]

加载优先级： 1️. 初始化参数 2️. 系统环境变量 3️. .env 文件 4️. YAML 文件 5️. TOML 文件 6️. 文件机密目录（如 /var/run/secrets）

maize.CrawlerProcess

Python
CrawlerProcess(
    settings: Optional[SpiderSettings] = None,
    settings_path: str | None = "settings.Settings",
)

运行爬虫

__get_settings staticmethod

Python
__get_settings(settings_path: str | None) -> SpiderSettings

获取配置 :param settings_path: :return:

crawl async

Python
crawl(
    spider: Union[
        Type[StandardSpiderInterface],
        StandardSpiderInterface,
    ],
)

装配爬虫 :param spider: Spider类 :return:

start async

Python
start()

开始运行所有爬虫 :return:

maize.BasePipeline

Python
BasePipeline(settings: SpiderSettings)

close abstractmethod async

Python
close()

管道关闭时调用，需要关闭的异步方法请在此实现 @return:

open abstractmethod async

Python
open()

管道初始化时调用，需要初始化的异步方法请在此实现 @return:

process_error_item abstractmethod async

Python
process_error_item(items: list[Item])

处理超过重试次数的数据 :param items: @return:

process_item abstractmethod async

Python
process_item(items: list[Item]) -> bool

处理数据，需要处理数据的方法请在此实现。 为了提高效率，请使用异步方法。 :param items: @return:

maize.BaseDownloader

Python
BaseDownloader(crawler: Crawler)

Bases: ABC

download abstractmethod async

Python
download(
    request: Request,
) -> Union[DownloadResponse, Request]

下载

:param request: 请求实例 :return: 返回 DownloadResponse 实例或 Request 实例

process_error_request async

Python
process_error_request(request: Request)

处理超过最大重试次数的请求 :param request: :return:

random_wait async

Python
random_wait()

随机等待

:return: