分解高度分支解析器的最pythonic方法

Question

我正在为特定类型的文件开发解析器，该文件由一些标题关键字分解为多个部分，然后是一堆异构数据。标题总是用空行分隔。大致如下：

Header_A

1 1.02345
2 2.97959
...

Header_B

1   5.1700   10.2500
2   5.0660   10.5000
...

每个标题都包含非常不同类型的数据，并且根据块中的某些关键字，数据必须存储在不同的位置。我采用的一般方法是使用一些正则表达式来捕获所有可以定义标题的关键字，然后遍历文件中的行。找到匹配项后，我会弹出行，直到到达空白行，将行中的所有数据存储在适当的位置。

这是代码的基本结构，其中“使用 current_line 执行操作”将涉及一堆分支，具体取决于行包含的内容：

headers = re.compile(r"""
    ((?P<header_a>Header_A)
    |
    (?P<header_b>Header_B))
    """, re.VERBOSE)

i = 0
while i < len(data_lines):
    match = header.match(data_lines[i])
    if match:
        if match.group('header_a'):
            data_lines.pop(i)
            data_lines.pop(i)

            #     not end of file         not blank line
            while i < len(data_lines) and data_lines[i].strip():
                current_line = data_lines.pop(i)
                # do stuff with current_line

        elif match.group('header_b'):
            data_lines.pop(i)
            data_lines.pop(i)

            while i < len(data_lines) and data_lines[i].strip():
                current_line = data_lines.pop(i)
                # do stuff with current_line
        else:
            i += 1
    else:
        i += 1

一切正常，但它相当于一个高度分支的结构，我发现它非常难以辨认，对于不熟悉代码的人来说可能很难理解。这也使得将行保持在

我正在做的一件事是将每个标题的分支分成单独的函数。这有望大大提高可读性，但是...

...有没有更简洁的方法来执行外部循环/匹配结构？也许使用 itertools？

此外，由于各种原因，此代码必须能够在 2.7 中运行。

Answer 1

您可以使用itertools.groupby 根据您希望执行的处理功能对行进行分组：

import itertools as IT

def process_a(lines):
    for line in lines:
        line = line.strip()
        if not line: continue        
        print('processing A: {}'.format(line))

def process_b(lines):
    for line in lines:
        line = line.strip()
        if not line: continue        
        print('processing B: {}'.format(line))

def header_func(line):
    if line.startswith('Header_A'):
        return process_a
    elif line.startswith('Header_B'):
        return process_b
    else: return None  # you could omit this, but it might be nice to be explicit

with open('data', 'r') as f:
    for key, lines in IT.groupby(f, key=header_func):
        if key is None:
            if func is not None:
                func(lines)
        else:
            func = key

应用于您发布的数据，上面的代码打印

processing A: 1 1.02345
processing A: 2 2.97959
processing A: ...
processing B: 1   5.1700   10.2500
processing B: 2   5.0660   10.5000
processing B: ...

---

上面代码中复杂的一行是

for key, lines in IT.groupby(f, key=header_func):

让我们试着把它分解成它的组成部分：

In [31]: f = open('data')

In [32]: list(IT.groupby(f, key=header_func))
Out[32]: 
[(<function __main__.process_a>, <itertools._grouper at 0xa0efecc>),
 (None, <itertools._grouper at 0xa0ef7cc>),
 (<function __main__.process_b>, <itertools._grouper at 0xa0eff0c>),
 (None, <itertools._grouper at 0xa0ef84c>)]

IT.groupby(f, key=header_func) 返回一个迭代器。迭代器产生的项目是 2-tuples，例如

(<function __main__.process_a>, <itertools._grouper at 0xa0efecc>)

二元组中的第一项是header_func 返回的值。 2 元组中的第二项是迭代器。这个迭代器从f 产生行，header_func(line) 都返回相同的值。

因此，IT.groupby 正在根据header_func 的返回值对f 中的行进行分组。当f 中的行是标题行时——Header_A 或Header_B——然后header_func 返回process_a 或process_b，我们希望使用该函数来处理后续行。

当f 中的行是标题行时，IT.groupby（2 元组中的第二项）返回的行组很短且无趣——它只是标题行。

我们需要在下一组中寻找有趣的行。对于这些行，header_func 返回None。

所以我们需要查看两个 2 元组：IT.groupby 产生的第一个 2 元组为我们提供了要使用的函数，第二个 2 元组给出了应该应用标头函数的行。

一旦你有了有趣的行的函数和迭代器，你只需调用func(lines)就可以了！

请注意，扩展它以处理其他类型的标头非常容易。您只需要编写另一个process_* 函数，并在line 指示时修改header_func 以返回process_*。

---

编辑：我删除了 izip(*[iterator]*2) 的使用，因为 它假定第一行是标题行。第一行可能是空白行或非标题行，这会使所有内容都失败。我用一些if-statements 替换了它。它没有那么简洁，但结果更可靠。

Answer 2

如何将解析不同标头的数据类型的逻辑拆分为单独的函数，然后使用字典从给定的标头映射到正确的标头：

def parse_data_a(iterator):
    next(iterator) # throw away the blank line after the header
    for line in iterator:
        if not line.strip():
            break  # bale out if we find a blank line, another header is about to start
        # do stuff with each line here

# define similar functions to parse other blocks of data, e.g. parse_data_b()

# define a mapping from header strings to the functions that parse the following data
parser_for_header = {"Header_A": parse_data_a} # put other parsers in here too!

def parse(lines):
    iterator = iter(lines)
    for line in iterator:
        header = line.strip()
        if header in parser_for_header:
            parser_for_header[header](iterator)

此代码使用迭代而不是索引来处理行。这样做的一个优点是除了在行列表之外，您还可以直接在文件上运行它，因为文件是可迭代的。它还使边界检查变得非常容易，因为当迭代中没有任何内容时，for 循环将自动结束，以及当break 语句被命中时。

根据您对正在解析的数据所做的工作，您可能需要让各个解析器返回一些内容，而不是只做他们自己的事情。在这种情况下，您需要在顶级 parse 函数中使用一些逻辑来获取结​​果并将其组合成一些有用的格式。也许字典最有意义，最后一行变成：

results_dict[header] = parser_for_header[header](iterator)

Answer 3

您也可以使用生成器的send 函数来实现：)

data_lines = [
    'Header_A   ',
    '',
    '',
    '1 1.02345',
    '2 2.97959',
    '',
]

def process_header_a(line):
    while True:
        line = yield line
        # process line
        print 'A', line

header_processors = {
    'Header_A': process_header_a(None),
}

current_processer = None
for line in data_lines:
    line = line.strip()
    if line in header_processors:
        current_processor = header_processors[line]
        current_processor.send(None)
    elif line:
        current_processor.send(line)    

for processor in header_processors.values():
    processor.close()

如果替换，您可以从主循环中删除所有if 条件

current_processer = None
for line in data_lines:
    line = line.strip()
    if line in header_processors:
        current_processor = header_processors[line]
        current_processor.send(None)
    elif line:
        current_processor.send(line)    

与

map(next, header_processors.values())
current_processor = header_processors['Header_A']
for line in data_lines:
    line = line.strip()
    current_processor = header_processors.get(line, current_processor)
    line and line not in header_processors and current_processor.send(line)