Python中Apache日志的Lexer

Question

最近，我用的书学习了Python，当我学习一种新的语言时，我正在做一些我的katas。这有助于获得“语言如何工作”，我为Apache日志编写了一个简单的lexer。

日志记录的一个示例是

64.242.88.10 - - [07/Mar/2004:16:05:49 -0800] "GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1" 401 12846
64.242.88.10 - - [07/Mar/2004:16:06:51 -0800] "GET /twiki/bin/rdiff/TWiki/NewUserTemplate?rev1=1.3&rev2=1.2 HTTP/1.1" 200 4523
64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291

我编写的程序将所有值读入标记，并使用|作为分隔符输出它们，如

64.242.88.10|-|-|07/Mar/2004:16:05:49 -0800|GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1|401|12846
64.242.88.10|-|-|07/Mar/2004:16:06:51 -0800|GET /twiki/bin/rdiff/TWiki/NewUserTemplate?rev1=1.3&rev2=1.2 HTTP/1.1|200|4523
64.242.88.10|-|-|07/Mar/2004:16:10:02 -0800|GET /mailman/listinfo/hsdivision HTTP/1.1|200|6291

不过，该lexer可以用于更广泛的使用。

这是我的代码，我很想看看你对如何改进它的评论和建议。我还就某些问题提出了几个具体问题。

import collections
import io
import sys

LEXER_SKIP_WHITESPACE=0
LEXER_READ_QUOTED_STRING=1
LEXER_READ_BRACKETED_STRING=2
LEXER_READ_WORD=3

class Location:
    def __init__(self, name=None, pos=0, line=1, col=0):
        self.name = name or "<input>"
        self.line = line
        self.col = col
        self.pos = pos
    def update(self, c):
        self.pos += 1
        if c == "\n":
            self.line += 1
            self.col = 0
        else:
            self.col += 1
    def __repr__(self):
        return str.format("Location({}, {}, {}, {})", repr(self.name), repr(self.pos), repr(self.line), repr(self.col))
    def __str__(self):
        return str.format("{}: {}: line {}, column {}", self.name, self.pos, self.line, self.col)

def readchar(inputchannel, location):
    while True:
        maybechar = inputchannel.read(1)
        if maybechar == '':
            return None
        else:
            location.update(maybechar)
            yield maybechar

def readtoken(inputchannel, location):
    state = LEXER_SKIP_WHITESPACE
    token = ''
    for nextchar in readchar(inputchannel, location):
        if state is LEXER_SKIP_WHITESPACE:
            if nextchar == "\n":
                yield "\n"
                continue
            elif nextchar.isspace():
                continue
            elif nextchar == '"':
                state = LEXER_READ_QUOTED_STRING
                continue
            elif nextchar == '[':
                state = LEXER_READ_BRACKETED_STRING
                continue
            else:
                state = LEXER_READ_WORD
                token += nextchar
                continue
        elif state is LEXER_READ_QUOTED_STRING:
            if nextchar == '"':
                yield token
                token = ''
                state = LEXER_SKIP_WHITESPACE
                continue
            else:
                token += nextchar
                continue
        elif state is LEXER_READ_BRACKETED_STRING:
            if nextchar == ']':
                yield token
                token = ''
                state = LEXER_SKIP_WHITESPACE
                continue
            else:
                token += nextchar
                continue
        elif state is LEXER_READ_WORD:
            if nextchar == "\n":
                yield token
                token = ''
                state = LEXER_SKIP_WHITESPACE
                yield "\n"
                continue
            elif nextchar.isspace():
                yield token
                token = ''
                state = LEXER_SKIP_WHITESPACE
                continue
            else:
                token += nextchar
                continue
        else:
            raise Error("Impossible lexer state.")
    if state is LEXER_SKIP_WHITESPACE:
        return None
    elif state is LEXER_READ_QUOTED_STRING:
        raise Error("End of character stream in quoted string.")
    elif state is LEXER_READ_BRACKETED_STRING:
        raise Error("End of character stream in quoted string.")
    elif state is LEXER_READ_WORD:
        yield token
        return None
    else:
        raise Error("Impossible lexer state.")


class Lexer:
    def __init__(self, inputchannel, _location=None):
        self.location = _location or Location("<input>", 0, 1, 0)
        self.inputchannel = inputchannel
        self.buf = ''
        self.state = LEXER_SKIP_WHITESPACE

    def __iter__(self):
        return readtoken(self.inputchannel, self.location)

if __name__ == "__main__":
    sep = ''
    for token in Lexer(sys.stdin):
        if token == '\n':
            sys.stdout.write(token)
            sep = ''
        else:
            sys.stdout.write(sep)
            sys.stdout.write(token)
            sep = '|'

问题1-如何用Python编写词汇者？

我很清楚有些类似于Python的lex和yacc的工具可以在这里使用，但这将使学习如何用Python编写这样的程序的目的落空。我发现这令人吃惊的困难。

我的第一个不幸之处是Python不做尾消除，因此基本上禁止将lexer编写为一组相互递归的函数。交互递归函数是我最喜欢的工具之一，因为它清楚地指出了lexer (它所处的递归函数)的一个特定状态，以及从此状态到其他状态的转换，并使每个转换都易于测试用例。

由于可能不是每个人都熟悉基于相互递归函数的词汇，所以这里的内容相当于OCaml中的readtoken生成器。read_token的开头是“如果您读取双引号，放弃它并执行read_quotedstring”，该函数本身在后面定义，并执行on期望的操作:它将缓冲区中的字符聚合到下一个双引号，并作为令牌祝福结果。

let rec read_token f ((ax,buffer) as state) s =
  match Stream.peek s with
  | Some('"') -> Stream.junk s; read_quotedstring f state s
  | Some('[') -> Stream.junk s; read_timestamp f state s
  | Some(' ')
  | Some('\t')-> Stream.junk s; read_token f state s
  | Some(c) -> read_simpleword f state s
  | None -> ax
and read_simpleword f ((ax,buffer) as state) s =
  match Stream.peek s with
  | Some('"')
  | Some('[')
  | Some(' ')
  | Some('\t') ->
    let current_token = Buffer.contents buffer in
    Buffer.clear buffer;
    read_token f (f ax current_token, buffer) s
  | Some(c) ->
    Buffer.add_char buffer c;
    Stream.junk s;
    read_simpleword f state s
  | None ->
    ax
and read_quotedstring f ((ax,buffer) as state) s =
  match Stream.peek(s) with
  | Some('"') ->
    Stream.junk s;
    let current_token = Buffer.contents buffer in
    Buffer.clear buffer;
    read_token f (f ax current_token, buffer) s
  | Some(c) ->
    Stream.junk s;
    Buffer.add_char buffer c;
    read_quotedstring f state s
  | None ->
    failwith "End of stream within a quoted string"
and read_timestamp f ((ax,buffer) as state) s =
  match Stream.peek(s) with
  | Some(']') ->
    Stream.junk s;
    let current_token = Buffer.contents buffer in
    Buffer.clear buffer;
    read_token f (f ax (timestamp_to_iso current_token), buffer) s
  | Some(c) ->
    Stream.junk s;
    Buffer.add_char buffer c;
    read_timestamp f state s
  | None ->
    failwith "End of stream within a bracketed string"

我的Python版本将几个状态定义为符号常量，就像我在C中所做的那样，然后构建一个巨大的while循环，跟踪转换。我对这个实现并不满意，因为它没有给我任何工具来管理lexer的复杂性。使用函数是非常不愉快的，因为Python中变量范围的细节。那么，如果我决定编写一个更复杂的解析器，那么在小的可测试部分中分解lexer的惯用方法是什么呢？用对象来表示词汇状态的想法会有趣吗？

问题2.将stdin视为字符流是正确的吗？

显然，我以某种方式做到了这一点，但是Python没有真正的字符类型，并且使它们看起来像长度为1的字符串。在我看来，将输入块读入“循环可扩展缓冲区”并复制这些块的子字符串以生成标记的方法在语言中更适合。我说的对吗？

问题3. Python中的通用异常是什么？

这似乎是一个相当基本的问题，但我确实未能在文献资料中找到合适的答案-- Exception似乎是一个糟糕的选择，因为它非常笼统，使得错误识别和错误处理相当复杂。

问题4.生成器不返回任何内容吗？

在生成器中返回None是一种好的方式吗？会通过也行吗？

Answer 1

除了其他答案之外，我还想在Q1和Q4上添加我自己的想法。

问题1

将readtoken分解为多个函数.

现在，您正在使用单个递归函数中的状态机来尝试读取不同类型的令牌。正如在另一个答案中提到的，递归并不是非常"pythonic“。此外，由于每个递归中的状态都会发生变化，因此很难跟踪此函数中的逻辑。

如果将不同的状态分解成单独的函数，这个逻辑将更容易理解，就像在OCaml示例中所做的那样。您甚至可以使用类似的高级函数名。然后，我们可以根据从流中读取的第一个字符，将readtoken的定义更改为路由到其他辅助函数。

def readtoken(inputchannel, location):
    for nextchar in readchar(inputchannel, location):
        if nextchar == '\n':
            yield '\n'
        elif nextchar.isspace():
            continue
        elif nextchar == '"':
            yield read_quoted_string(inputchannel, location)
        elif nextchar == '[':
            yield read_bracketed_string(inputchannel, location)
        else:
            yield read_token(nextchar, inputchannel, location)

注意函数是如何立即变得更容易阅读的。我们将我们的职责委托给其他命名函数，就像在您的OCaml示例中一样。readtoken的目的现在只是一个路由器，在这里，我们根据从流中读取的第一个字符将工作委托给其他函数。

现在，每个帮助函数都有一个单独的责任--从流中读取它们给定的令牌类型，并返回结果。注意，这些函数实际上应该是return结果，而不是yield函数--这些函数不是生成器，这里唯一的生成器是readtoken函数。

助手函数实现：

def read_quoted_string(inputchannel, location):
    token = ''
    for nextchar in readchar(inputchannel, location):
        if next_char == '"':
            return token
        else:
            token += nextchar
    raise Exception("End of character stream in quoted string.")


def read_bracketed_string(inputchannel, location):
    token = ''
    for nextchar in readchar(inputchannel, location):
        if nextchar == ']':
            return token
        else:
            token += nextchar
    raise Exception("End of character stream in bracketed string.")


def read_token(token, inputchannel, location):
    for nextchar in readchar(inputchannel, location):
        if nextchar.isspace():
            return token
        else:
            token += nextchar
    return token # if we reach the end of the stream

请注意以下几点：

• 每个函数的可读性都大大提高。我们不再需要通过一个递归函数来跟踪它所处的状态。
• 现在我们可以看到，read_quoted_string和read_bracketed_string有很多共同的逻辑。也许我们应该将它们组合成一个函数read_string，并以一个string_delimiter作为参数。
• read_token使用一个初始令牌作为它的第一个参数--这是我们从顶级readtokens函数中读取的字符。如果我们不喜欢这个实现，我们可以尝试在顶层使用peek来解决它。

不使readtoken成为生成器函数。

在您的示例中，Lexer类实现了__iter__，它将其视为可迭代的。这意味着您的readtoken不需要是生成器函数(即您可以用return替换所有yield语句)，因为Lexer类已经是一个环绕它的生成器。

问题4

当生成器函数退出时，它会自动引发StopIteration。for-loops和while-loops将通过结束循环并继续运行来自动处理这个StopIteration。

因此，结束生成器的“正常”方式是简单地结束该函数。return None确实做到了这一点，但更传统的方法是简单地到达函数的末尾并自然返回。您可以通过在您的break函数中从while-loop中提取readchar来实现这一点。