在Python中处理文本搜索和CSV输出的需求时,核心目标是减少不必要的遍历次数,同时保证结果的准确性和完整性。我们可以先把所有要搜索的字符串整理成集合,再逐行读取文本文件,每行同时匹配所有目标字符串,这样能大幅提升处理效率。

实现思路梳理
整个流程可以分为三个核心步骤:
- 准备待搜索的字符串列表,转换为集合提升查找效率
- 逐行读取文本文件,记录当前行号,同时检查该行是否包含任意目标字符串
- 将匹配到的行号、行内容、匹配到的字符串等信息写入CSV文件
完整实现代码
基础版本实现
以下是完整的可运行代码,包含异常处理和结果输出逻辑:
import csv
def search_strings_in_text_to_csv(text_file_path, target_strings, output_csv_path):
"""
在文本文件中搜索多个字符串,将匹配结果写入CSV
:param text_file_path: 待搜索的文本文件路径
:param target_strings: 要搜索的字符串列表
:param output_csv_path: 输出CSV文件路径
"""
# 将目标字符串转为集合,提升in操作效率
target_set = set(target_strings)
# 存储匹配结果
match_results = []
try:
with open(text_file_path, 'r', encoding='utf-8') as text_file:
line_num = 1
for line in text_file:
# 去除行尾换行符
line_content = line.rstrip('n')
# 检查当前行是否包含任意目标字符串
matched_strings = [s for s in target_set if s in line_content]
if matched_strings:
# 记录匹配结果,每个匹配字符串单独存一条记录
for matched_str in matched_strings:
match_results.append({
'line_number': line_num,
'line_content': line_content,
'matched_string': matched_str
})
line_num += 1
except FileNotFoundError:
print(f"错误:未找到文本文件 {text_file_path}")
return
except Exception as e:
print(f"读取文本文件时发生错误:{e}")
return
# 将结果写入CSV文件
try:
with open(output_csv_path, 'w', encoding='utf-8', newline='') as csv_file:
fieldnames = ['line_number', 'line_content', 'matched_string']
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(match_results)
print(f"搜索完成,共找到 {len(match_results)} 条匹配记录,结果已写入 {output_csv_path}")
except Exception as e:
print(f"写入CSV文件时发生错误:{e}")
# 示例调用
if __name__ == "__main__":
# 待搜索的字符串列表
search_strings = ["error", "warning", "failed"]
# 文本文件路径
text_path = "test_log.txt"
# 输出CSV路径
csv_path = "search_result.csv"
search_strings_in_text_to_csv(text_path, search_strings, csv_path)
代码逻辑说明
代码中首先将目标字符串列表转为set类型,因为Python中集合的in操作时间复杂度是O(1),比列表的O(n)效率高很多,尤其是目标字符串数量较多时优势明显。逐行读取文本文件时,我们用列表推导式快速筛选出当前行包含的所有目标字符串,避免多次遍历目标集合。最后使用csv.DictWriter将结果写入CSV,保证格式规范,不会出现内容包含逗号导致格式错乱的问题。
性能优化技巧
如果处理的文本文件体积非常大,比如几个G的日志文件,还可以做以下优化:
- 使用
with语句读取文件,避免文件句柄泄露,同时逐行读取不会一次性把整个文件加载到内存,适合大文件处理 - 如果目标字符串有固定的前缀或者模式,可以先用
re.compile预编译正则表达式,再使用正则匹配,比多次in判断效率更高 - 如果不需要重复匹配同一个字符串多次,可以在匹配到之后从目标集合中临时移除该字符串,减少后续匹配的判断次数
正则匹配优化版本
当目标字符串包含特殊字符或者需要模糊匹配时,可以使用正则提升灵活性:
import csv
import re
def search_strings_regex_to_csv(text_file_path, target_patterns, output_csv_path):
"""
使用正则表达式搜索文本文件中的多个模式,结果写入CSV
:param text_file_path: 待搜索的文本文件路径
:param target_patterns: 正则表达式模式列表
:param output_csv_path: 输出CSV文件路径
"""
# 预编译所有正则模式
compiled_patterns = [re.compile(pattern) for pattern in target_patterns]
match_results = []
try:
with open(text_file_path, 'r', encoding='utf-8') as text_file:
line_num = 1
for line in text_file:
line_content = line.rstrip('n')
for pattern in compiled_patterns:
if pattern.search(line_content):
match_results.append({
'line_number': line_num,
'line_content': line_content,
'matched_pattern': pattern.pattern
})
line_num += 1
except FileNotFoundError:
print(f"错误:未找到文本文件 {text_file_path}")
return
except Exception as e:
print(f"处理文件时发生错误:{e}")
return
try:
with open(output_csv_path, 'w', encoding='utf-8', newline='') as csv_file:
fieldnames = ['line_number', 'line_content', 'matched_pattern']
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(match_results)
print(f"搜索完成,共找到 {len(match_results)} 条匹配记录,结果已写入 {output_csv_path}")
except Exception as e:
print(f"写入CSV文件时发生错误:{e}")
# 示例调用,搜索以error开头或者包含404的行
if __name__ == "__main__":
patterns = [r"^error", r"404"]
search_strings_regex_to_csv("test_log.txt", patterns, "regex_result.csv")
这个版本适合需要复杂匹配规则的场景,比如匹配固定格式的错误码、特定开头的日志行等,预编译正则可以减少重复编译的开销,提升整体运行效率。