我是怎么开发一个数据提取轮子 DataExtractor
Mar 16, 2021
8 minute read

前言

数据提取对于爬虫开发来说可占了不少开发时间,是其中及其重要的一部分。 为了降低这方面的时间和心智成本,我调研且使用了不少轮子。

  • 像使用 XPath 作为选择器的 XSLT 1,对应 Python 的轮子就是最经典且好用的 lxml 2
  • 像在 Scrapy 中使用的的 Selectors 3,来自 Parsel 4;
  • 像 Ruia 内置的 Data Item 5
  • 像到现在刚从 Scrapy 中重构独立出来的 Item Loaders 6 ,来自 itemloaders 7

这些轮子有的很久远,比如 XSLT 1,只支持文本 to 文本的转换。

其间 Scrapy 的 Parsel 支持用 CSS Selectors 8 及 XPath 9 去对 HTML(XML) 10 文本数据进行提取。 而 itemloaders 通过更高层的抽象,基于 Parsel 定义提取器类,去对文本进行复杂的数据提取, 支持嵌套提取(不支持类定以嵌套,支持函数调用式去提取),还支持对原始数据进行格式转换等处理。

有的很新颖,像 Ruia Data Item 5 支持编写提取器类,再把对应文本转换成类对象。(但该提取器的表达能力没有 itemloaders 的好)

在 19 年的时候,我自己开发了一个数据提取的轮子 DataExtractor 11。 那时候,这些轮子不是使用太复杂;就是提供的接口太 low level 了,而没法提高工作效率;或者无法处理更多场景,比如 JSON 文本数据。 其中的 ruia 虽然刚开发没多久,但其 Data Item 的点子对我开发 DataExtractor 有很重要的参考价值。

理想中轮子的模样

当时理想中轮子的模样如下,

  • 支持多种提取场景,支持提取 HTML(XML) 10 及 JSON 12 数据。
  • 支持定义提取器类。
  • 支持提取器嵌套提取。
  • 支持多种提取表达式,例如:XPath,XPath2.0 9,CSS Selectors 8 ,JSONPath 13 等。
  • 支持对提取结果进行处理,格式转换等操作。

要支持这么多功能,到底要怎么开发呢?

实现

过了太久了,当时磕磕碰碰怎么写的,其实我也记不住了(论及时记录的好处)…… 现在得重新看一下源码,把思路梳理整理一下。

数据格式及其查询表达式

作为一个爬虫工程师,经常面对的文本数据的数据格式一般为 HTML 10 及 JSON 12 格式。 如今,已经有很多 DSL 方便我们去处理这些数据类型。 其中用来查询定位该数据结构中的某个节点的 DSL,我一般称之为查询表达式 Query Expression(有些复杂到都可以称作是 Language 了)。

对于不同数据格式,所广泛使用的查询表达式标准有

HTML 10

  • XPath 9

    全称为 XML Path Language。 十分古老,诞生于 98 年;且功能强大,17 年还发布了版本 3.1 的标准;而且用来起来是十分方便。通过以下例子,简单的说明一下如何通过 XPath 去定位 XML 文件数据结构中,不同的数据节点。

    用浏览器打开链接 https://www.rssboard.org/files/sample-rss-2.xml 打开一个 RSS 文件(格式为 XML)。 呼出开发者工具(FireFox 快捷键为 Shift + CMD + C ),通过 XPath 表达式 //channel/title 即可方便地定位到这个 RSS 文件的频道名称。 通过表达式 //channel/item/title 方便地获取到 RSS 文件中所有的文章标题。

  • CSS Selectors 8

    CSS 全称是层叠样式表( Cascading Style Sheets)。CSS Selectors 本来是用来标记哪些节点的样式,让浏览器知道如何去渲染,比如渲染字体颜色,图片大小等。 比如,以下代码就是用来把所有 class 带 bar 的 h1 节点的字体渲染成绿色。

    h1.bar { color: green }  /* h1 elements with class~=bar */

    由于 CSS Selectors 用起来也挺方便的,且简短,也经常用来定位数据。 回到刚 RSS 文件的例子,通过 CSS Selectors 表达式 channel > title 也可以定位到这个 RSS 文件的频道名称

JSON 12

  • JSONPath 13

    关于这个的标准有很多(比如我造的轮子 jsonpath-extractor),但最早的标准应该来自于这一篇博客 JSONPath - XPath for JSON | stefan.goessner

    而现在运用最多且最广为人知的应该是 jq,是一个十分好用的命令行工具。 不过我经常用来美观命令行输出的 JSON 内容,像这样 cat response.json | jq . -C | less

以上的数据场景和对应的查询表达式,对应到 Python 的轮子有好几个。但为了方便阐述,我们以以下三个轮子为例子

  • lxml 用来支持 HTML 文本的解析,及支持 XPath 表达式 9
  • cssselect 用来支持 CSS Selectors 表达式 8(实际是把 CSS Selectors 表达式转换成相同语义的 XPath 表达式)
  • jsonpath-extractor 来支持 JSONPath 表达式(这个与前面提到的 JSONPath 的博客 13 的设计不同)。 而 JSON 文本的解析由标准库 json 来支持

而这些轮子该怎么用呢,请看以下两个例子。

  • 解析刚刚的用到的 RSS 文件,再提取出所有文章标题。

    import requests
    from lxml.etree import fromstring
        
    rss_file_url = "http://www.rssboard.org/files/sample-rss-2.xml"
    resp = requests.get(rss_file_url)
        
    root = fromstring(resp.text)
    xpath_expr = "//channel/item/title/text()"
    print(f"xpath: {xpath_expr!r} result:\n", root.xpath(xpath_expr))
        
    cssselect_expr = "channel > item > title"
    print(
        f"cssselect: {cssselect_expr!r} result:\n",
        [el.text for el in root.cssselect(cssselect_expr)],
    )
        
    from cssselect import GenericTranslator
    from lxml.etree import XPath
        
    sel = XPath(GenericTranslator().css_to_xpath(cssselect_expr))
    print("css2xpath result:", sel)

    代码执行结果

    xpath: '//channel/item/title/text()' result:
     ['Star City', 'The Engine That Does More', "Astronauts' Dirty Laundry"]
    cssselect: 'channel > item > title' result:
     ['Star City', 'The Engine That Does More', "Astronauts' Dirty Laundry"]
    css2xpath result: descendant-or-self::channel/item/title

    从 lxml 文档中关于 cssselect 的说明,了解到 CSS Selectors 实际上是转换成 XPath 表达式,再用 XPath 的接口去执行查询的。 感兴趣的话,可以去看一下 lxml 中相关的源码。顺便提一下,其源码绝大部分是 cython 写的。

  • 解析 JSON 文本,再通过 JSONPath 表达式提取所需要的数据。

    import json
        
    from jsonpath import parse
        
    raw = """{
        "goods": [
            {"price": 100, "category": "Comic book"},
            {"price": 200, "category": "magazine"},
            {"price": 200, "no category": ""}
        ],
        "targetCategory": "book"
    }"""
        
    data = json.loads(raw)
        
        
    print("get all prices:\n", parse("$.goods[*].price").find(data))
    print(
        "query goods which price is lower than 150:\n",
        parse("$.goods[@.price < 150]").find(data),
    )
    print(
        "more complex:\n",
        parse("$.goods[contains(@.category, $.targetCategory)]").find(data),
    )

    代码执行结果

    get all prices:
     [100, 200, 200]
    query goods which price is lower than 150:
     [{'price': 100, 'category': 'Comic book'}]
    more complex:
     [{'price': 100, 'category': 'Comic book'}]

实现提取器,适配不同的场景

从以上两个场景,我们可以找出其共同点及不同点。

  • 共同点
    1. 输入一个查询表达式字符串,一个数据源
    2. 输出一个查询结果列表
  • 不同点
    1. 数据源的类型不同。HTML 文本需经过 lxml.etree.fromstring 去做解析成树结构;JSON 文本也得经过 json.loads 去做解析成树结构。 且两者树结构的类型不同,结构也不同。
    2. 输出查询结果列表中元素的数据类型不同。不都是字符串,比如用 CSS Selectors 表达式提取后,还需对其结果做进一步处理,才能拿到标题节点的内容。

为了兼容不同场景下的使用,求同存异,我定义了一个抽象类(对应文件 data_extractor/core.py )。

from abc import abstractmethod
from typing import Any, List

sentinel = object()


class ExprError(Exception):
    def __init__(self, extractor: "AbstractSimpleExtractor", exc: Exception):
        self.extractor = extractor
        self.exc = exc

    def __str__(self) -> str:
        return (
            f"ExprError with {self.exc!r} "
            f"raised by {self.extractor!r} extracting"
        )

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}({self.extractor!r}, exc={self.exc!r})"
        )


class AbstractSimpleExtractor:
    def __init__(self, expr: str):
        self.expr = expr

    @abstractmethod
    def extract(self, element: Any) -> List[Any]:
        raise NotImplementedError

    def extract_first(self, element: Any, default: Any = sentinel) -> Any:
        rv = self.extract(element)
        if not rv:
            if default is sentinel:
                raise ExprError(self, element)

            return default

        return rv[0]

通过继承抽象类 AbstractSimpleExtractor 实现不同场景下的提取器类。 统一提取的接口,只对外暴露 extractextract_first 方法, 包括提取数据过程的所有可能产生的异常也以 ExprError 包装后抛出。隐藏了具体的实现细节。

  • extract 返回查询结果列表
  • extract_first 返回列表里的第一个查询结果。如果没有,则使用默认值; 未配置默认值,则抛出 ExprError 异常

现以 XPath 及 CSS Selectors 为例子,定义提取器 (对应文件 data_extractor/lxml.py )。

from typing import List, Union

from cssselect import GenericTranslator
from cssselect.parser import SelectorError
from lxml.etree import XPath, XPathEvalError, XPathSyntaxError
from lxml.etree import _Element as Element

from .core import AbstractSimpleExtractor, ExprError


class XPathExtractor(AbstractSimpleExtractor):
    def __init__(self, expr: str):
        super().__init__(expr)
        try:
            self._find = XPath(self.expr)
        except XPathSyntaxError as exc:
            raise ExprError(extractor=self, exc=exc) from exc

    def extract(self, element: Element) -> Union[List[Element], List[str]]:
        try:
            rv = self._find(element)
            if not isinstance(rv, list):
                # normalize-space XPath function return value type is str
                return [rv]
            else:
                return rv
        except XPathEvalError as exc:
            raise ExprError(extractor=self, exc=exc) from exc


class CSSExtractor(AbstractSimpleExtractor):
    def __init__(self, expr: str):
        super().__init__(expr)
        try:
            xpath_expr = GenericTranslator().css_to_xpath(self.expr)
        except SelectorError as exc:
            raise ExprError(extractor=self, exc=exc) from exc

        self._extractor = XPathExtractor(xpath_expr)

    def extract(self, element: Element) -> List[Element]:
        return self._extractor.extract(element)


class TextCSSExtractor(CSSExtractor):
    def extract(self, element: Element) -> List[str]:
        return [ele.text for ele in super().extract(element)]

由于 CSSExtractor 的结果只能为 Element 类实例,不是我们想要的字符串格式。 所以,需要实现一个 TextCSSExtractor 来调用 Element.text 属性来获取节点的文本数据。 不像 XPathExtractor 可以直接修改 XPath 表达式来获取 text 属性的数据。 例如: //channel/item/title 只能获取到文章的标题元素,修改成 //channel/item/title/text() 即可获取其文本内容。

那现在以同样的场景来测试一下,统一接口后的提取器模块。

import requests
from lxml.etree import fromstring

from data_extractor.lxml import TextCSSExtractor, XPathExtractor

rss_file_url = "http://www.rssboard.org/files/sample-rss-2.xml"
resp = requests.get(rss_file_url)

root = fromstring(resp.text)
xpath_expr = "//channel/item/title/text()"
print(
    f"xpath: {xpath_expr!r} result:\n", XPathExtractor(xpath_expr).extract(root)
)

cssselect_expr = "channel > item > title"
print(
    f"cssselect: {cssselect_expr!r} result:\n",
    TextCSSExtractor(cssselect_expr).extract(root),
)

代码执行结果

xpath: '//channel/item/title/text()' result:
 ['Star City', 'The Engine That Does More', "Astronauts' Dirty Laundry"]
cssselect: 'channel > item > title' result:
 ['Star City', 'The Engine That Does More', "Astronauts' Dirty Laundry"]

复杂的数据提取场景

以上只是统一了提取器的接口,通过继承实现抽象类,即可适配了各个场景下的数据提取。 但目前这个轮子还没达到提高生产力的作用,因为实际的提取场景更为复杂,不是简单提取几个数据就够了。

还是以刚刚的 RSS 文件为例子,真实的爬取场景下,单单提取各个文章的标题肯定是不够的,还可能需要文章的链接、摘要、发布时间等数据。

因为 XML 格式是树结构的,数据具有层次结构,有规律地嵌套着的。这样通过 //channel/item 能查询到所有 item 节点。遍历节点列表以以下查询表达式,分别获取各个文章所需要的数据。

  • ./title/text() 标题
  • ./link/text() 链接
  • ./description/text() 摘要
  • ./pubDate/text() 发布时间

那我们赶紧用刚实现的模块,去做数据提取。看看效果怎么样

import pprint

import requests
from lxml.etree import fromstring

from data_extractor.core import ExprError
from data_extractor.lxml import XPathExtractor

rss_file_url = "http://www.rssboard.org/files/sample-rss-2.xml"
resp = requests.get(rss_file_url)

root = fromstring(resp.text)

articles = []
X = XPathExtractor  # shorten the class name
for node in X("//channel/item").extract(root):
    article = {
        "title": X("./title/text()").extract_first(node, default=""),
        "link": X("./link/text()").extract_first(node, default=""),
        "desc": X("./description/text()").extract_first(node, default=""),
        "published": X("./pubDate/text()").extract_first(node, default=None),
    }
    articles.append(article)


print("extracted result:")
pprint.pprint(articles[:2], width=80)  # beautify extracted result

代码运行结果

extracted result:
[{'desc': 'How do Americans get ready to work with Russians aboard the '
          'International Space Station? They take a crash course in culture, '
          "language and protocol at Russia's <a "
          'href="http://howe.iki.rssi.ru/GCTC/gctc_e.htm">Star City</a>.',
  'link': 'http://liftoff.msfc.nasa.gov/news/2003/news-starcity.asp',
  'published': 'Tue, 03 Jun 2003 09:39:21 GMT',
  'title': 'Star City'},
 {'desc': 'Sky watchers in Europe, Asia, and parts of Alaska and Canada will '
          'experience a <a '
          'href="http://science.nasa.gov/headlines/y2003/30may_solareclipse.htm">partial '
          'eclipse of the Sun</a> on Saturday, May 31st.',
  'link': '',
  'published': 'Fri, 30 May 2003 11:06:42 GMT',
  'title': ''}]

这代码看起来还是写起来挺繁琐的。 为了提高工作效率(实际为了少写点代码 ( ;´Д`)),必须引入新的抽象,消灭重复代码!

通过刚刚的例子,很容易就能想出新的抽象的输入为,一个节点提取器,多个子提取器,及这些提取器的执行行为,默认值。 为了符合数据的结构,新的抽象也需支持嵌套,能一层层的对数据做提取。还要方便复用及修改。

再整理一下,新的抽象所需要支持的功能

  • 输入
    • 节点提取器
    • 多个子提取器
    • 提取器的执行行为,默认值等
  • 支持嵌套组合提取
  • 支持复用及修改

为了方便复用,节点提取器需经常被修改;相对于它,子提取器需要修改的情况较少。 实际的应用场景也是这样的,同样的数据结构,在整个数据里可能会出现在多个地方。 比如,一个新闻网站上的文章摘要卡片就会出现在推荐栏,滚动横窗等位置。

现在我们来确认一下这个抽象的接口。

  1. 通过继承类,子提取器定义为类变量的方式,去定义复杂提取器

    这样做的好处是可以继承去复用修改,添加新的子提取器,或者覆盖掉原来的。 通过元类编程的方式去实现,这样就可以在类定义的时候,收集到哪些类变量是子提取器。

  2. 一个用来提取数据的方法 extract

    不像刚刚的定义的模块有两个方法 extractextract_first ,由于提取行为是预先定义的,所以一个方法就够了。

  3. 修改比较频繁的参数像节点定位器,提取行为及默认值,都以实例化传参的方式进行配置。方便修改

以下就是新的抽象(对应文件 data_extractor/item.py

from typing import Any, Dict, Iterator, Tuple

from .core import AbstractSimpleExtractor, ExprError, sentinel


def is_simple_extractor(obj: Any) -> bool:
    return isinstance(obj, AbstractSimpleExtractor)


class ComplexExtractorMeta(type):
    def __init__(
        cls,  # noqa: B902
        name: str,
        bases: Tuple[type],
        attr_dict: Dict[str, Any],
    ):
        super().__init__(name, bases, attr_dict)

        field_names = []
        for key, attr in attr_dict.items():
            if isinstance(type(attr), ComplexExtractorMeta):
                field_names.append(key)

        field_names.extend(getattr(cls, "_field_names", []))
        cls._field_names = field_names


class Field(metaclass=ComplexExtractorMeta):
    def __init__(
        self,
        extractor: AbstractSimpleExtractor = None,
        name: str = None,
        default: Any = sentinel,
        is_many: bool = False,
    ):
        if extractor is not None and not is_simple_extractor(extractor):
            raise ValueError(f"Invalid SimpleExtractor: {extractor!r}")

        if default is not sentinel and is_many:
            raise ValueError(
                f"Can't both set default={default} and is_many=True"
            )

        self.extractor = extractor
        self.name = name
        self.default = default
        self.is_many = is_many

    def __repr__(self) -> str:
        args = [f"{self.extractor!r}"]
        if self.name is not None:
            args.append(f"name={self.name!r}")

        if self.default is not sentinel:
            args.append(f"default={self.default!r}")

        if self.is_many:
            args.append(f"is_many={self.is_many!r}")

        return f"{self.__class__.__name__}({', '.join(args)})"

    def extract(self, element: Any) -> Any:
        rv = self.extractor.extract(element)

        if self.is_many:
            return [self._extract(r) for r in rv]

        if not rv:
            if self.default is sentinel:
                raise ExprError(self, element)

            return self.default

        return self._extract(rv[0])

    def _extract(self, element: Any) -> Any:
        return element


class Item(Field):
    def _extract(self, element: Any) -> Any:
        rv = {}
        for field in self.field_names():
            try:
                extractor = getattr(self, field)
                if extractor.name is not None:
                    field = extractor.name

                rv[field] = extractor.extract(element)
            except ExprError as exc:
                raise exc

        return rv

    @classmethod
    def field_names(cls) -> Iterator[str]:
        yield from cls._field_names

现在用这个新模块再试验一下,看看使用起来感觉怎么样?

import pprint

import requests
from lxml.etree import fromstring

from data_extractor.item import Field as F
from data_extractor.item import Item
from data_extractor.lxml import XPathExtractor as X

rss_file_url = "http://www.rssboard.org/files/sample-rss-2.xml"
resp = requests.get(rss_file_url)
root = fromstring(resp.text)


class Article(Item):
    title = F(X("./title/text()"), default="")
    link = F(X("./link/text()"), default="")
    desc = F(X("./description/text()"), default="")
    published = F(X("./pubDate/text()"), default=None)


articles = Article(X("//channel/item"), is_many=True).extract(root)
print("extracted result:")
pprint.pprint(articles[:2], width=80)  # beautify extracted result

代码运行结果如下

extracted result:
[{'desc': 'How do Americans get ready to work with Russians aboard the '
          'International Space Station? They take a crash course in culture, '
          "language and protocol at Russia's <a "
          'href="http://howe.iki.rssi.ru/GCTC/gctc_e.htm">Star City</a>.',
  'link': 'http://liftoff.msfc.nasa.gov/news/2003/news-starcity.asp',
  'published': 'Tue, 03 Jun 2003 09:39:21 GMT',
  'title': 'Star City'},
 {'desc': 'Sky watchers in Europe, Asia, and parts of Alaska and Canada will '
          'experience a <a '
          'href="http://science.nasa.gov/headlines/y2003/30may_solareclipse.htm">partial '
          'eclipse of the Sun</a> on Saturday, May 31st.',
  'link': '',
  'published': 'Fri, 30 May 2003 11:06:42 GMT',
  'title': ''}]

写到这里,DataExtractor 的核心功能已经讲完了。作为一个小轮子,这功能已经足够在各个场景下使用了。 除了爬虫,在其他地方也有其他用处。比如数据清理,可以通过 DataExtractor 配合 JSONPath 可以迅速地对 JSON 数据进行结构转换,快速的清理数据。

轮子们功能对比

功能 DataExtractor Scrapy Itemloaders Ruia Data Item
支持 XPath ✔️ ✔️ ✔️
支持 CSS Selectors ✔️ ✔️ ✔️
支持 JSONPath ✔️
支持 Regex ✔️
支持嵌套组合提取 ✔️ ✔️
支持提取后的数据处理 ✔️ ✔️
支持对提取结果的类型注解 ❌ ️

那肯定有人想问,牺牲空闲时间实现这个有啥意义呢?其他的轮子又不是不能用,至于自己实现么。 没错,对比其他轮子所实现的功能,看似没什么竞争力……但这种事情属于见仁见智了。

除了功能,评价轮子好不好,也得看使用者的上手速度快不快;写出来的代码的可阅读程度好不好;可扩展性高不高等。 但这些因素比较主观呐,对使用者接受程度也有一定的要求。

展望

近来随着 Python 圈子中类型注解 Type Hinting 及数据类 Data Classes 的火热,在使用这两者的过程中,我也积攒了一些想法。 下一步开发计划就是,使 DataExtractor 支持更细粒度的类型注解。(预计又要水一篇博客了,也可能得等个两年 233) 实现这个功能可能要实现一个 mypy 插件 14,及用 TypedDict 15 来作为复杂提取器提取结果的默认的类型注解。

Footnotes

1 XSLT | wikipedia

2 XSLT | lxml Docs

3 Selectors | Scrapy Docs

4 Parsel Docs

5 Define Data Item | Ruia Docs

6 Item Loaders | Scrapy Docs

7 itemloaders Docs

8 CSS selectors | MDN

9 XPath | MDN

10 HTML | wikipedia XML | wikipedia

11 linw1995/[email protected]

12 JSON | wikipedia

13 JSONPath - XPath for JSON | stefan.goessner

14 Extending and integrating mypy

15 PEP 589 – TypedDict: Type Hints for Dictionaries with a Fixed Set of Keys




comments powered by Disqus