我有一個可迭代的bytes，例如
bytes_iter = (
b'col_1,',
b'c',
b'ol_2\n1',
b',"val',
b'ue"\n',
)
（但通常這不會被硬編碼或一次全部可用，而是由生成器提供）并且我想將其轉換為可迭代的str行，其中換行符在前面是未知的，但可以是\r,\n或\r\n. 所以在這種情況下將是：
lines_iter = (
'col_1,col_2',
'1,"value"',
)
（但同樣，就像一個可迭代的，而不是一次全部在記憶體中）。
我怎樣才能做到這一點？
背景關系：我的目標是然后將 str 行的可迭代傳遞給csv.reader（我認為需要整行？），但我一般對這個答案感興趣。
uj5u.com熱心網友回復：
使用該io模塊為您完成大部分作業：
class ReadableIterator(io.IOBase):
def __init__(self, it):
self.it = iter(it)
def read(self, n):
# ignore argument, nobody actually cares
# note that it is *critical* that we suppress the `StopIteration` here
return next(self.it, b'')
def readable(self):
return True
然后打電話io.TextIOWrapper(ReadableIterator(some_iterable_of_bytes))。
uj5u.com熱心網友回復：
我用yield和re.finditer。
yield 運算式在定義生成器函式或異步生成器函式時使用，因此只能在函式定義的主體中使用。在函式體中使用 yield 運算式會導致該函式成為生成器函式
回傳一個迭代器，該迭代器在字串中 RE 模式的所有非重疊匹配中產生匹配物件。從左到右掃描字串，并按找到的順序回傳匹配項。結果中包含空匹配項。
如果沒有組，則回傳與整個模式匹配的字串串列。如果只有一個組，則回傳與該組匹配的字串串列。如果存在多個組，則回傳與組匹配的字串元組串列。非捕獲組不影響結果的形式。
正則運算式([^\r\n]*)(\r\n|\r|\n)?可以分為兩部分進行匹配（即兩組）。第一組匹配沒有\rand的資料\n，第二組匹配\r, \nor \r\n。
import re
find_rule = re.compile("([^\r\n]*)(\r\n|\r|\n)?")
def converter(byte_data):
left_d = ""
for d in byte_data:
# Used to save the previous match result in the `for` loop
prev_result = None
# Concatenate the last part of the previous data with the current data,
# used to deal with the case of `\r\n` being separated.
d = left_d d.decode()
left_d = ""
# Using `find_rule.finditer` the last value("") will be invalid
for match_result in find_rule.finditer(d):
i = match_result.group()
if not i:
# The program comes to this point, indicating that i == "", which is the last matching value
left_d, prev_result = prev_result.group(), None
continue
if prev_result:
if prev_result.group(2) is None:
# The program goes here, represented as the last valid value matched
left_d = prev_result.group()
else:
# Returns the previous matched value
yield prev_result.group()
# Save the current match result
prev_result = match_result
else:
yield left_d
for i in (converter(iter((
b'col_1,\r',
b'\nc',
b'ol_2\n1',
b'\n,"val;\r',
b'ue"\n')))
):
print(repr(i))
輸出：
'col_1,\r\n'
'col_2\n'
'1\n'
',"val;\r'
'ue"\n'
uj5u.com熱心網友回復：
也許我錯過了一些重要（或微妙）的東西，因為一些被贊成的答案似乎比這更奇特，但我認為你可以解碼和鏈接位元組并使用itertools.groupby來獲取字串生成器：
from itertools import groupby, chain
bytes_iter = (
b'col_1,',
b'c',
b'ol_2\n',
b'1,"val;',
b'ue"\n'
)
def make_strings(G):
strings = chain.from_iterable(map(bytes.decode, G))
for k, g in groupby(strings, key=lambda c: c not in '\n\r'):
if k:
yield ''.join(g)
list(make_strings(bytes_iter))
# ['col_1,col_2', '1,"val;ue"']
uj5u.com熱心網友回復：
將 @o11c 和 @user2357112 放在一起支持 Monica 的貢獻：
import codecs
import csv
import io
def yield_bytes():
chunks = [
b'col_1,',
b'c',
b'ol_2\n1',
b',"val',
b'ue"\n',
b'Hello,'
b'\xe4\xb8',
b'\x96',
b'\xe7',
b'\x95\x8c\n'
b'\n'
]
for chunk in chunks:
yield(chunk)
decoder = codecs.getincrementaldecoder('utf-8')()
def yield_encoded_bytes():
s = None
for bytes in yield_bytes():
s = decoder.decode(bytes, final=False)
if s:
yield s.encode('utf-8')
class ReadableIterator(io.IOBase):
def __init__(self, it):
self.it = iter(it)
def read(self, n):
# ignore argument, nobody actually cares
# note that it is *critical* that we suppress the `StopIteration` here
return next(self.it, b'')
def readable(self):
return True
f = io.TextIOWrapper(ReadableIterator(yield_encoded_bytes()))
for row in csv.reader(f):
print(row)
我得到：
['col_1', 'col_2']
['1', 'value']
['Hello', '世界']
[]
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/410008.html
標籤：
上一篇：csv.DictWriterwriteheader()和writerow在MicrosoftExcel中顯示時無法在CSV檔案中正確寫入希臘字符