Python 字符串操作——性能问题

Question

我有以下代码，我在我的应用程序中执行了大约 200 万次来解析这么多记录。这部分似乎是瓶颈，我想知道是否有人可以通过提出一些可以使这些简单的字符串操作更快的技巧来帮助我。

try:
    data = []
    start = 0
    end = 0
    for info in self.Columns():
        end = start + (info.columnLength)
        slice = line[start:end]
        if slice == '' or len(slice) != info.columnLength:
            raise 'Wrong Input'
        if info.hasSignage:
            if(slice[0:1].strip() != '+' and slice[0:1].strip() != '-'):
                raise 'Wrong Input'
        if not info.skipColumn:
            data.append(slice)
        start = end 
    parsedLine = data
except:
    parsedLine = False

Answer 1

编辑：我正在稍微改变这个答案。我会在下面留下原来的答案。

在我的另一个答案中，我评论说最好的办法是找到一个内置的 Python 模块来进行解包。我想不出一个，但也许我应该让谷歌搜索一个。 @John Machin 提供了一个答案，展示了如何做到这一点：使用 Python struct 模块。由于它是用 C 编写的，它应该比我的纯 Python 解决方案更快。 （我实际上没有测量任何东西，所以这是一个猜测。）

我同意原始代码中的逻辑是“非 Pythonic”。返回哨兵值不是最好的；最好返回一个有效值或引发异常。另一种方法是返回一个有效值列表以及另一个无效值列表。由于@John Machin 提供了生成有效值的代码，我想我会在这里编写一个返回两个列表的版本。

注意：也许最好的答案是采用@John Machin 的答案并对其进行修改以将无效值保存到文件中以供以后查看。他的答案一次只产生一个答案，因此无需构建大量已解析记录；并且将坏行保存到磁盘意味着不需要构建一个可能很大的坏行列表。

import struct

def parse_records(self):
    """
    returns a tuple: (good, bad)
    good is a list of valid records (as tuples)
    bad is a list of tuples: (line_num, line, err)
    """

    cols = self.Columns()
    unpack_fmt = ""
    sign_checks = []
    start = 0
    for colx, info in enumerate(cols, 1):
        clen = info.columnLength
        if clen < 1:
            raise ValueError("Column %d: Bad columnLength %r" % (colx, clen))
        if info.skipColumn:
            unpack_fmt += str(clen) + "x"
        else:
            unpack_fmt += str(clen) + "s"
            if info.hasSignage:
                sign_checks.append(start)
        start += clen
    expected_len = start
    unpack = struct.Struct(unpack_fmt).unpack

    good = []
    bad = []
    for line_num, line in enumerate(self.whatever_the_list_of_lines_is, 1):
        if len(line) != expected_len:
            bad.append((line_num, line, "bad length"))
            continue
        if not all(line[i] in '+-' for i in sign_checks):
            bad.append((line_num, line, "sign check failed"))
            continue
        good.append(unpack(line))

    return good, bad

原始答案文本： 如果self.Columns() 信息在所有记录中都相同，那么这个答案应该会快很多。我们一次性处理self.Columns() 信息，并构建了几个列表，其中包含我们处理记录所需的内容。

这段代码展示了如何计算parsedList，但实际上并没有产生它或返​​回它或用它做任何事情。显然你需要改变它。

def parse_records(self):
    cols = self.Columns()

    slices = []
    sign_checks = []
    start = 0
    for info in cols:
        if info.columnLength < 1:
            raise ValueError, "bad columnLength"
        end = start + info.columnLength
        if not info.skipColumn:
            tup = (start, end)
            slices.append(tup)   
            if info.hasSignage:
                sign_checks.append(start)

    expected_len = end # or use (end - 1) to not count a newline

    try:
        for line in self.whatever_the_list_of_lines_is:
            if len(line) != expected_len:
                raise ValueError, "wrong length"
            if not all(line[i] in '+-' for i in sign_checks):
                raise ValueError, "wrong input"
            parsedLine = [line[s:e] for s, e in slices]

    except ValueError:
        parsedLine = False

Answer 2

def fubarise(data):
    try:
        if nasty(data):
            raise ValueError("Look, Ma, I'm doing a big fat GOTO ...") # sheesh #1
        more_of_the_same()
        parsed_line = data
    except ValueError:
        parsed_line = False
        # so it can be a "data" or False -- sheesh #2
    return parsed_line

raise 语句中有不同的错误消息是没有意义的；他们从未见过。 Sheesh #3。

更新：这是一个改进建议，它使用struct.unpack 快速划分输入行。它还说明了更好的异常处理，假设代码的编写者也在运行它并且在第一个错误时停止是可以接受的。为用户受众记录所有行的所有列中的所有错误的健壮实现是另一回事。请注意，通常每列的错误检查会更广泛，例如检查前导符号但不检查列是否包含有效数字似乎有点奇怪。

import struct

def unpacked_records(self):
    cols = self.Columns()
    unpack_fmt = ""
    sign_checks = []
    start = 0
    for colx, info in enumerate(cols, 1):
        clen = info.columnLength
        if clen < 1:
            raise ValueError("Column %d: Bad columnLength %r" % (colx, clen))
        if info.skipColumn:
            unpack_fmt += str(clen) + "x"
        else:
            unpack_fmt += str(clen) + "s"
            if info.hasSignage:
                sign_checks.append(start)
        start += clen
    expected_len = start
    unpack = struct.Struct(unpack_fmt).unpack

    for linex, line in enumerate(self.whatever_the_list_of_lines_is, 1):
        if len(line) != expected_len:
            raise ValueError(
                "Line %d: Actual length %d, expected %d"
                % (linex, len(line), expected_len))
        if not all(line[i] in '+-' for i in sign_checks):
            raise ValueError("Line %d: At least one column fails sign check" % linex)
        yield unpack(line) # a tuple

Answer 3

怎么样（使用一些类有一个可执行的例子）：

class Info(object):
    columnLength = 5
    hasSignage = True
    skipColumn = False

class Something(object):

    def Columns(self):
        return [Info()]*4

    def bottleneck(self):
        try:
            data = []
            start = 0
            end = 0
            line = '+this-is just a line for testing'
            for info in self.Columns():
                start = end
                collength = info.columnLength
                end = start + collength
                if info.skipColumn:  # start with this
                    continue

                elif collength == 0: 
                    raise ValueError('Wrong Input')

                slice = line[start:end] # only now slicing, because it
                                        # is probably most expensive part

                if len(slice) != collength: 
                    raise ValueError('Wrong Input')

                elif info.hasSignage and slice[0] not in '+-': # bit more compact
                    raise ValueError('Wrong Input')

                else:
                    data.append(slice)

            parsedLine = data
        except:
            parsedLine = False

Something().bottleneck()

编辑： 当切片长度为0时，切片[0]不存在，所以必须先检查if collength == 0

编辑2： 您将这段代码用于许多行，但列信息不会改变，对吧？这样你就可以

• 预先计算每列的起点列表（不再需要计算起点、终点）
• 提前知道 start-end，.Columns() 只需要返回未跳过且列长度 >0 的列（或者您真的需要在每一行都为 length==0 提高输入吗？？）
• 每行的强制长度已知且相等或每行可以在遍历列信息之前检查

编辑3： 我想知道如果您使用'skipColumn'，您将如何知道哪个数据索引属于哪个列...

Answer 4

我想告诉你使用某种内置的 Python 功能来拆分字符串，但我想不出一个。所以我只剩下试图减少你拥有的代码量了。

当我们完成后，end 应该指向字符串的末尾；如果是这种情况，那么所有的.columnLength 值一定是没问题的。 （除非是负面的什么的！）

由于 this 引用了 self，它必须是来自成员函数的片段。因此，您可以return False 提前退出函数并返回错误标志，而不是引发异常。但我喜欢将 except 子句更改为不再捕获异常并获取堆栈跟踪以帮助您确定问题出在哪里的调试潜力。

@Remi 使用了slice[0] in '+-'，而我使用了slice.startswith(('+', '-))。我想我更喜欢@Remi 的代码，但我保持不变只是为了向您展示不同的方式。 .startswith() 方法适用于长度超过 1 的字符串，但由于这只是长度为 1 的字符串，因此简洁的解决方案有效。

try:
    line = line.strip('\n')
    data = []
    start = 0
    for info in self.Columns():
        end = start + info.columnLength
        slice = line[start:end]
        if info.hasSignage and not slice.startswith(('+', '-')):
            raise ValueError, "wrong input"
        if not info.skipColumn:
            data.append(slice)
        start = end

    if end - 1 != len(line):
        raise ValueError, "bad .columnLength"

    parsedLine = data

except ValueError:
    parsedLine = False

Answer 5

我首先要考虑的是slice = line[start:end]。切片创建新实例；您可以尝试避免显式构造 line [start:end] 并手动检查其内容。

你为什么要slice[0:1]？这应该产生一个包含slice 的单个项目的子序列（不应该吗？），因此它可能可以更有效地检查。

Answer 6

不要在每次循环中都计算start 和end。

在使用 self.Columns() 之前只计算一次（不管是什么。如果 'Columns' 是具有静态值的类，那很愚蠢。如果它是一个名称以大写字母开头的函数，那就令人困惑了。）

if slice == '' or len(slice) != info.columnLength 仅在行与Columns 所需的总大小相比太短时才会发生。在循环外检查一次。

slice[0:1].strip() != '+' 确实看起来像 .startswith()。

if not info.skipColumn。在开始循环之前应用此过滤器。从self.Columns() 中删除这些。