Python 编写文件解析器

Python 编写文件解析器，可以将文件解析看作归约。许多语言使用了双层定义：语言的底层标记，以及建于其上的高级结构。以XML文件为例，标签、标签名称以及属性名称构成了底层语法，XML描述的整体结构构成了高级语法。

底层的文本扫描实际上是对单个字符的归约，并把它们分组为标记，用Python的生成器函数设计模式可以轻松实现。常规的实现方法如下：

def lexical_scan(some_source):
    for char in some_source:
        if some pattern completed: yield token
        else: accumulate token

可以使用底层文件解析器来完成这部分工作，使用CSV、JSON和XML库处理细节，我们会基于这些库开发高级解析器。

沿用双层设计模式，底层解析器生成原始数据的正式数据结构，即由文本元组组成的迭代器，兼容多种格式的数据文件。高级解析器则负责生成可供应用程序使用的对象，可以是数字元组、命名元组或者其他类型的Python不可变对象。

前面用到了一种底层解析器，其输入是KML文件，KML文件是XML表达地理位置信息的一种格式。解析器的核心功能如下：

def comma_split(text: str) -> List[str]:
    return text.split(",")
def row_iter_kml(file_obj: TextIO) -> Iterator[List[str]]:
    ns_map = {
        "ns0": "http://www.opengis.net/kml/2.2",
        "ns1": "http://www.google.com/kml/ext/2.2"}
    xpath = (
        "./ns0:Document/ns0:Folder/"
        "ns0:Placemark/ns0:Point/ns0:coordinates")
    doc = XML.parse(file_obj)
    return (
        comma_split(cast(str, coordinates.text))
        for coordinates in doc.findall(xpath, ns_map)
    )

row_iter_kml()函数解析XML文件的方法是通过doc.findall()函数遍历文档中所有的<ns0:coordinates>标签，然后通过comma_split()函数将标签中的文本转换为三元组。

cast()函数的作用是为mypy标示出coordinates.text是字符串类型值，text属性的默认类型是Union[str, bytes]，但这里只有字符串这种情况。cast()不执行任何运行时操作，仅用作mypy的类型标示。

该解析器可以处理标准化的XML结构。输入文档必须满足数据库设计中第一范式的要求，即每个属性必须满足原子化和单一值的要求，XML数据的每一行拥有相同的列，每列中元素的类型一致。数据的值并非完全原子化的，需要用逗号分隔为经度、纬度和海拔高度的原子字符串。但文本类型完全一致，符合第一范式。

将大量数据（包括XML标签、属性以及其他标记符号）归约为相对少量的信息，即浮点数类型的经度和纬度，所以解析器实际上进行的是某种归约计算。

现在需要将元组中的字符串转换为浮点数，去掉海拔高度，调整经纬度顺序，从而得到符合应用程序要求的元组格式，具体变换过程如下所示：

def pick_lat_lon(
        lon: Any, lat: Any, alt: Any) -> Tuple[Any, Any]:
    return lat, lon

def float_lat_lon(
        row_iter: Iterator[Tuple[str, ...]]
    ) -> Iterator[Tuple[float, ...]]:
    return (
        tuple(
            map(float, pick_lat_lon(*row))
        )
        for row in row_iter
    )

关键所在的float_lat_lon()是高阶函数，返回生成器表达式。其中生成器使用map()函数将float()函数应用于pick_lat_lon()函数的返回结果，参数*row将行数据元组的每个元素取出作为pick_lat_lon()函数的各个参数，所以只有每行为三元组时才可用。pick_lat_lon()函数再对输入值进行筛选和重新排序，返回符合要求的元组。

如下所示使用解析器：

name = "file:./Winter%202012-2013.kml"
with urllib.request.urlopen(name) as source:
    trip = tuple(float_lat_lon(row_iter_kml(source)))

返回结果是从原来KML文件中转换来的路径上的一系列路径点，每个点采用嵌套元组的形式。使用底层解析器从原始数据中抽取行文本，再使用高级解析器将文本转换为浮点数元组，这里没有执行任何验证。

解析CSV文件

前面介绍了另一个解析非标准CSV文件的例子。当时为了去掉文件头，使用了一个简单的函数提取文件头，然后返回了包含剩余行数据的迭代器。

原始数据如下所示：

Anscombe's quartet
I   II    III     IV
x   y    x   y    x    y   x   y
10.0     8.04     10.0     9.14    10.0   7.46     8.0     6.58
8.0     6.95     8.0    8.14    8.0   6.77   8.0      5.76
...
5.0     5.68     5.0    4.74    5.0   5.73   8.0      6.89

数据之间以Tab字符分隔，文件前3行是需要去掉的文件头部分。

下面是CSV解析器的另一个版本，我们通过3个函数来实现。第一个函数row_iter()返回包含Tab分隔符的迭代器，如下所示：

def row_iter_csv(source: TextIO):
    rdr= csv.reader(source, delimiter="\t")
    return rdr

只是简单地包装了CSV文件解析器，之前实现过的XML和普通文本解析器并不包含这一部分。解析器处理标准格式数据的常规方法是创建基于行元组的迭代器。

得到行元组后，就可以保留包含可用数据的行，去掉包含其他元数据的行，例如标题和列名称。这里需要引入一个用于解析的辅助函数，以及用于验证行数据的过滤函数。

转换函数如下所示：

from typing import Optional, Text
def float_none(data: Optional[Text]) -> Optional[float]:
    try:
        data_f = float(data)
        return data_f
    except ValueError:
        return None

该函数负责将格式正确的字符串转换为浮点数，将无效数据转换为None。类型标示Optional[Text]和Optional[float]表示对应值是符合类型定义的普通值或者None值。

下面把该函数嵌入到一个映射中，从而将整行数据转换为浮点数或者None值，该匿名函数如下所示：

from typing import Callable, List, Optional
R_Text = List[Optional[Text]]
R_Float = List[Optional[float]]

float_row: Callable[[R_Text], R_Float] \
    = lambda row: list(map(float_none, row))

两个类型标示显式定义了float_row。R_Text定义了行数据的文本版本是个包含了文本值和None值的列表。R_Float定义了行数据的实数版本。

下面是一个行级验证器，使用all()函数确保所有值都是浮点数（或者说没有None值）：

all_numeric: Callable[[R_Float], bool] \
    = lambda row: all(row) and len(row) == 8

该匿名函数属于归约函数，将一行实数归约为一个布尔值。如果这些数据都不为“假”（既不是None值，也不是零值），并且个数正好是8，则返回布尔“真”值。

上面这个简化版的all_numeric函数不区分零值和None值。更准确的测试条件应该类似于not any(item is None for item in row)，具体实现留给读者作为练习。

该函数设计的核心目标是创建基于行的元素，并将其整合到完整算法中以解析输入文件。基本函数迭代处理文本元组，联合起来对结果数据进行转换及验证。对于满足第一范式（所有行相同）或者可以通过简单方法排除异常数据行的情况，该函数表现不错。

但大多数解析工作不会这么简单。有些文件的头部或者尾部包含重要数据，虽然与其他部分的格式不同，却不能简单地丢弃，需要用更复杂的解析器处理这类非标准文件。

解析带文件头的普通文本文件

前面有个没有经过解析处理的Crayola.GPL文件，如下所示：

GIMP Palette
Name: Crayola
Columns: 16
#
239 222 205   Almond
205 149 117   Antique Brass

可以使用正则表达解析文本文件。首先用一个过滤器读取并解析文件头，然后返回一个包含数据行的可迭代序列。这个复杂的二步解析器完全是基于由两部分（文件头和尾）组成的文件结构定义的。

如下底层解析器可以处理包含4行文件头和大量尾部数据行：

Head_Body = Tuple[Tuple[str, str], Iterator[List[str]]]
def row_iter_gpl(file_obj: TextIO) -> Head_Body:
    header_pat = re.compile(
        r"GIMP Palette\nName:\s*(.*?)\nColumns:\s*(.*?)\n#\n",
        re.M)

    def read_head(
            file_obj: TextIO
        ) -> Tuple[Tuple[str, str], TextIO]:
        match = header_pat.match(
            "".join(file_obj.readline() for _ in range(4))
        )
        return (match.group(1), match.group(2)), file_obj

    def read_tail(
            headers: Tuple[str, str],
            file_obj: TextIO) -> Head_Body:
        return (
            headers,
            (next_line.split() for next_line in file_obj)
        )

    return read_tail(*read_head(file_obj))

Head_Body类型定义了行迭代器的整体实现目标，返回结果是个二元组。其中第一个元素也是一个二元组，包含从头部提取的信息，第二个元素是一个包含颜色信息的迭代器。下面的函数有两处用到了这个Head_Body类型定义。

header_pat是一个用于解析前4行文件头的正则表达式，表达式中的两个括号用于提取头部的名称和列数信息。

接下来定义了两个内部函数来解析文件的不同部分。read_head()函数解析文件头，具体过程是：首先读取文件的前4行，将其合并为一个字符串，然后用head_pat正则表达式进行解析，返回结果包含了文件头中的两部分数据，以及处理剩余行的迭代器。

一个函数返回迭代器作为另一个函数的输入参数，基于在函数间传递状态对象的设计思想，但它只能算作一个简化实现，因为read_tail()函数的输入参数正是read_head()函数的输出。
read_tail()函数解析文件的剩余行，按照GPL文件格式的定义来解析，基于空格符分割字符串。

　更多信息，可访问https://code.google.com/p/grafx2/issues/detail?id=518。

当把文件中的每行文本解析为一系列字符串元组后，就可以用高阶函数进行后续处理了，包括转换和验证（如若需要）。

高阶解析函数示例如下：

from typing import NamedTuple
class Color(NamedTuple):
    red: int
    blue: int
    green: int
    name: str

def color_palette(
        headers: Tuple[str, str],
        row_iter: Iterator[List[str]]
    ) -> Tuple[str, str, Tuple[Color, ...]]:
    name, columns = headers
    colors = tuple(
        Color(int(r), int(g), int(b), " ".join(name))
        for r, g, b, *name in row_iter)
    return name, columns, colors

该函数所需的头部数据与迭代器由底层函数row_iter_gpl()提供。使用多重赋值将颜色数据与剩余的名称分开，分别赋给4个变量r、g、b和name。使用*name形式的参数以确保所有剩余数据都会以元组的形式赋给保存名字的变量。最后" ".join(name)将这些单词用空格连接起来组成一个完整的字符串。

如下所示使用该双层解析器：

with open("crayola.gpl") as source:
    name, cols, colors = color_palette(
        *row_iter_gpl(source)
    )
print(name, cols, colors)

对底层函数的处理结果应用高阶函数，最终的返回结果是文件头数据和一系列Color对象组成的元组。