如何写一个简单的 DSL 从 JSON 中提取数据

Jul 21, 2020

3 minute read

前言

在进入正题之前，先简单地介绍下什么是 DSL(Domain Specific Language)。就是专用于特定领域的语言，比如 HTML，Markdown，Org，RST 等这些文本标记语言（CSS 也是 DSL，用来描述 HTML 的展示形式）；比如 Regex 来做文本匹配提取；比如 YAML，TOML，XML 这样的配置文件格式；比如 Graphviz 的 DOT 语言来描述图形信息； SQL 用来描述如何从数据库中获取数据。

与 DSL 相对应的就有 GPL(General Purpose Language) ，即通用的计算机编程语言。 GPL 是可以用来编写任意计算机程序的，并且能表达任何的可被计算的逻辑，同时也是图灵完备的语言。比如 Python，C，C++，Java 等编程语言。

使用一门优秀的 DSL 就可以真正的提升生产力，减少不必要的工作，在一些领域帮助我们更快的实现需求。

工作面临的痛点

写爬虫难免会处理各种各样的数据，其中最常见的就是 JSON 及 HTML 文本了。解析 HTML 并提取数据，我们有好用的轮子，比如 lxml ，及用来提取数据的 DSL 即 XPath（或者 CSS-Selectors ）。面对复杂的 JSON 嵌套结构，解析可以用标准库 json。而提取数据则没有好用的 DSL，所以我们难免会写出这样的代码

data = json.load(text)
username = data["info"]["author"]["username"]
blogs = []
for raw in data["info"]["blogs"]:
    blogs.append({
        "title": raw["header"]
        "content": raw["text"]
    })

为了避免字段丢失而抛出 KeyError 异常，我们常常会添加些而外的代码进行处理。比如套几个 try-except 代码块，代码就会变成这样

try:
    username = data["info"]["author"]["username"]
except KeyError:
    username = ""

blogs = []
try:
    for raw in data["info"]["blogs"]:
        try:
            blogs.append({
                "title": raw["header"]
                "content": raw["text"]
            })
        except KeyError:
            pass
except KeyError:
    pass

或者直接调用 dict.get 方法去获取，当字段不存在时，而使用缺省值。则代码会变成这样

username = data.get("info", {}).get("author", {}).get("username", "")
blogs = []
for raw in data.get("info", {}).get("blogs", []):
    blogs.append({
        "title": raw.get("header", "")
        "content": raw.get("text", "")
    })

明明就是只是想避免代码抛出 KeyError 异常，就把代码平白增添这么多，十分不符合 Python 哲学。也为了减少写重复的代码，所以我们得想办法把这段程序的代码写的简单点。

DSL 能解放生产力

通过实现一种 DSL 像 XPath 那样来从 JSON 数据中提取想要的数据。

JSON 数据的语法支持嵌套，如果平展开来，就像一层层的树状结构。从根部开始出发，经过一个个节点（节点却有不同的结构 key-value (object) 和 idx-value (array) ）最终到叶节点，也就是变量值（布尔值，Null，数字及字符串）。

我们通过字段名 key 当路径的名称，以半角句号隔开，来代表我们走过的一个个节点（一层层嵌套的对象），即可得到 "info.author.username"。这样我们就可以通过这简单的表达式，来定位我们想要的数据。

那怎么表示 JSON 的列表呢？以列表的下标来表示的话，我们可以写出 "info.blogs.0"，虽然能用，但这样又会与前面的表示形式起冲突。如果用中括号来表示列表，数字来表示下标的话，就能得到我们所熟悉的表达式 "info.blogs[0]"，常见于许多编程语言。

我就可以以这些表达式，来描述 JSON 里某个特定节点上的数据。比如以下代码，就是我理想中的实现。

username = parse("info.author.username").find(data)
blogs = parse("info.blogs[*]{title:header, content:text}").find(data)

如果用过 JQ 的人可能会对这些写法 "info.author.username" 及 "info.blogs[*]{title:header, content:text}" 感到熟悉。

接下来我们就来实现我梦里的 DSL(Domain-specific language 领域特定语言)。

实现一个 DSL 来从 JSON 中提取数据

只要简单三步即可

实现解析器 Parser，把 DSL 转换成 AST(Abstract Syntax Tree 抽象语法树)
实现可执行程序 Executable，把数据从 JSON 中提取出来
实现转换器 Transformer，把 AST 转成可执行可执行程序 (Code Generation)

实现解析器 Parser

这边用一个第三方开源库 lark-parser/lark 去实现。它基于 EBNF 实现了自己的扩展写法。支持多种解析算法。接下来用个例子来介绍一下，它是如何使用的。

from lark import Lark

parser = Lark('''?start: path
                 ?path: path "." name          -> chained_path
                     | name
                 name: CNAME

                 %import common.CNAME         -> CNAME
                 %import common.WS_INLINE

                 %ignore WS_INLINE
              ''')

tree = parser.parse("info.author.name")
print(tree)
print(tree.pretty())

运行以上代码，终端就会输出解析到的 AST 结果。

Tree(chained_path, [Tree(chained_path, [Tree(name, [Token(CNAME, 'info')]), Tree(name, [Token(CNAME, 'author')])]), Tree(name, [Token(CNAME, 'name')])])
chained_path
  chained_path
    name    info
    name    author
  name  name

实现执行程序 Executable

接下来要实现一个可执行的算法。目前只是为了举例，写的比较粗暴。实现一个高阶函数 getter 来创建相应的提取函数，在函数中先粗暴地拦截掉所有 KeyError 异常，返回成 None。再实现一个高阶函数 chain 来创建级联提取函数。

import operator

def getter(name):

    def _getter(data):
        try:
            return operator.getitem(data, name)
        except KeyError:
            return None

    return _getter

def chain(current, next):

    def pair(data):
        return next(current(data))

    return pair

实现转换器 Transformer

再实现个把 AST 转换成执行代码的转换器。只要自底向上从左到右遍历树结构（后序遍历），根据节点的树结构类型去执行对应的转换函数，构建我们的执行算法。比如 name 就只有一个子节点 CNAME；而 chained_path 就有两个子节点，表达式的历史路径，及表达式的下一个路径 name。

?path: path "." name -> chained_path
    | name
name: CNAME

虽然我们需要遍历树结构，但我们没有必要去实现相应的算法。这是因为 lark-parser 提供特别方便的 visitors 模块去让开发者使用。以类方法来转换 AST 上对应的结构。就是把 CNAME token 转换成字符串，把 name 结构转换成提取函数，把 chained_path 转换成级联（链式）提取函数。代码如下

from lark.visitors import Transformer, v_args

@v_args(inline=True)
class JSONPathTransformer(Transformer):
    CNAME = str

    def name(self, cname):
        return getter(cname)

    def chained_path(self, previous_path, name):
        return chain(previous_path, name)

transformer = JSONPathTransformer()

验证

接下来我们写几个简单的测试，测试一下

def parse(text):
    return transformer.transform(parser.parse(text))

assert parse("info.author.name")(
        {"info": {"author": {"name": "Jack"}}}
    ) == "Jack"
assert parse("data")({"info":"boo"}) is None
assert parse("info")({"info":"boo"}) == "boo

之前 Executable 部分的代码有个 bug ，不知道大家能不能一眼看出来。我给个提示，执行以下代码就会抛出异常，这又该怎么修复呢？

assert parse("info.author")({}) is None

答案就是 getter 得处理参数 data=None 的情况，不然以上代码就会抛出 TypeError 异常。

总结

以上只是个简单的例子来展示如何写个 DSL 来从 JSON 中提取数据，让大家对整个过程有个大概的了解。如果让大家自己来开发自己工作上想要的 DSL ，会是什么样的呢？是像 Regex 来做文本匹配提取，像 TOML， YAML 这样的配置文件格式，像 Graphviz 的 DOT 语言描述图形信息，像 SQL 用来描述如何从数据库中获取数据。大家以后可以开动一下脑洞，看看自己有领域上有什么工作可以用 DSL 来描述（当然创建一门 DSL 是来方便大家用的，而不是来给别人造麻烦的）。

下期预告如何实现生产可用的 DSL

当然以上那么点代码，如果用在工程上肯定是远远不够的。要做一个优秀的轮子，不能止于能用就行。还需要简单易懂的文档，大量的单元及功能测试，良好的代码设计，优秀的代码测试覆盖率，大量的代码注释等等。

如果大家觉得这个有趣，对这个感兴趣的话，下次我还可以再分享相关的内容。

衍生阅读

这篇文章只是粗浅的介绍了怎么实现一门描述如何从 JSON 中提取数据的 DSL。如果对这背后的原理感兴趣的话，可以去看下编译原理前端这一部分的内容。或者对接下来 JSON 提取表达式怎么实现感兴趣的话，可以去搜搜看学习一下别人的实现（想当初我也是参考了很多轮子）。大家可以想一下，怎么实现比较好？

谈谈 DSL 以及 DSL 的应用（以 CocoaPods 为例）
lark-parser/lark
JSON 的标准可以查看这个 RFC 7159 The JavaScript Object Notation (JSON) Data Interchange Format。
什么是 EBNF

linw1995

前言