本文目录导读：
我来介绍几种Python拆分大文本的方法,从简单到复杂：
方法1：按固定大小分割
def split_by_size(text, chunk_size=1000):
"""按固定字符数分割文本"""
return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
# 示例
large_text = "A" * 5000 # 模拟大文本
chunks = split_by_size(large_text, 1000)
print(f"分割成 {len(chunks)} 块")
for i, chunk in enumerate(chunks[:3]):
print(f"块{i+1}: 长度={len(chunk)}, 内容={chunk[:50]}...")
方法2：按行分割（文件处理）
def split_file_by_lines(filename, lines_per_chunk=1000):
"""按行数分割大文件"""
chunks = []
with open(filename, 'r', encoding='utf-8') as file:
current_chunk = []
for i, line in enumerate(file, 1):
current_chunk.append(line)
if i % lines_per_chunk == 0:
chunks.append(''.join(current_chunk))
current_chunk = []
# 处理最后剩余的行
if current_chunk:
chunks.append(''.join(current_chunk))
return chunks
# 示例使用
# chunks = split_file_by_lines('large_file.txt', 500)
方法3：按段落分割
def split_by_paragraph(text, max_paragraphs=5):
"""按段落分割文本"""
paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
chunks = []
for i in range(0, len(paragraphs), max_paragraphs):
chunk = '\n\n'.join(paragraphs[i:i+max_paragraphs])
chunks.append(chunk)
return chunks
# 示例
sample_text = """第一段内容...
..
..
..
.."""
paragraphs = split_by_paragraph(sample_text, 2)
print(f"分割成 {len(paragraphs)} 块")
方法4：智能分割（保留句子完整）
import re
def smart_split_by_size(text, max_size=500):
"""智能分割，尽量在句子结束处断开"""
if len(text) <= max_size:
return [text]
chunks = []
while len(text) > max_size:
# 在最大长度范围内找最后一个句子结束符
split_point = max_size
for delimiter in ['。', '！', '？', '\n', '.', '!', '?']:
last_delimiter = text[:max_size].rfind(delimiter)
if last_delimiter > 0:
split_point = last_delimiter + 1
break
chunks.append(text[:split_point])
text = text[split_point:]
if text:
chunks.append(text)
return chunks
# 示例
long_text = "这是一个测试文本。" * 100 + "最后一句。"
smart_chunks = smart_split_by_size(long_text, 500)
print(f"智能分割成 {len(smart_chunks)} 块")
方法5：使用生成器（内存友好）
def read_in_chunks(file_object, chunk_size=1024):
"""逐块读取文件，节省内存"""
while True:
data = file_object.read(chunk_size)
if not data:
break
yield data
# 示例
def process_large_file(filename):
with open(filename, 'r', encoding='utf-8') as f:
for i, chunk in enumerate(read_in_chunks(f, 1024)):
print(f"处理块 {i+1}: {len(chunk)} 字符")
# 在这里处理每个块
# process_chunk(chunk)
# process_large_file('very_large_file.txt')
实际应用示例
class TextSplitter:
"""文本分割器类"""
def __init__(self, method='size', chunk_size=1000, overlap=100):
self.method = method
self.chunk_size = chunk_size
self.overlap = overlap
def split(self, text):
if self.method == 'size':
return self._split_by_size(text)
elif self.method == 'paragraph':
return self._split_by_paragraph(text)
elif self.method == 'sentence':
return self._split_by_sentence(text)
else:
raise ValueError(f"Unknown method: {self.method}")
def _split_by_size(self, text):
"""带重叠的分割"""
chunks = []
start = 0
while start < len(text):
end = start + self.chunk_size
if end < len(text):
# 尝试在句子边界断开
for i in range(end, start, -1):
if text[i] in '.。!！?？\n':
end = i + 1
break
chunks.append(text[start:end])
start = end - self.overlap # 保留重叠部分
return chunks
def _split_by_paragraph(self, text):
paragraphs = text.split('\n\n')
chunks = []
current_chunk = []
current_size = 0
for para in paragraphs:
if current_size + len(para) > self.chunk_size and current_chunk:
chunks.append('\n\n'.join(current_chunk))
current_chunk = [para]
current_size = len(para)
else:
current_chunk.append(para)
current_size += len(para)
if current_chunk:
chunks.append('\n\n'.join(current_chunk))
return chunks
def _split_by_sentence(self, text):
"""按句子分割，然后组合成块"""
# 简单的句子分割
sentences = re.split(r'[。！？.!?]', text)
sentences = [s.strip() + '。' for s in sentences if s.strip()]
chunks = []
current_chunk = []
current_size = 0
for sentence in sentences:
if current_size + len(sentence) > self.chunk_size and current_chunk:
chunks.append(''.join(current_chunk))
current_chunk = [sentence]
current_size = len(sentence)
else:
current_chunk.append(sentence)
current_size += len(sentence)
if current_chunk:
chunks.append(''.join(current_chunk))
return chunks
# 使用示例
splitter = TextSplitter(method='sentence', chunk_size=500)
text = "这是第一句，这是第二句！这是第三句？这是第四句，这是第五句。" * 50
result = splitter.split(text)
print(f"分割成 {len(result)} 个文本块")
# 显示前3个块的信息
for i, chunk in enumerate(result[:3]):
print(f"块{i+1}: {len(chunk)}字符, 开头: {chunk[:50]}...")
性能优化建议
import time
from itertools import islice
# 1. 对于超大文件，使用迭代器模式
def efficient_split_large_file(filename, chunk_size_mb=10):
"""高效分割大文件，避免一次性加载到内存"""
chunk_size_bytes = chunk_size_mb * 1024 * 1024
with open(filename, 'rb') as f:
chunk_number = 0
while True:
chunk = f.read(chunk_size_bytes)
if not chunk:
break
chunk_number += 1
yield chunk_number, chunk
# 2. 并行处理（适用于多核CPU）
from multiprocessing import Pool
def process_chunk(chunk_data):
chunk_number, chunk = chunk_data
# 处理每个块
return f"Processed chunk {chunk_number}"
def parallel_split_process(filename, num_processes=4):
"""并行处理和分割"""
with Pool(num_processes) as pool:
results = pool.map(process_chunk,
efficient_split_large_file(filename))
return results
# 3. 性能对比测试
def benchmark_splitting(text, methods=['size', 'paragraph', 'sentence']):
"""测试不同分割方法的性能"""
for method in methods:
splitter = TextSplitter(method=method, chunk_size=500)
start_time = time.time()
chunks = splitter.split(text)
elapsed = time.time() - start_time
print(f"{method}方法: {elapsed*1000:.2f}ms, 生成{len(chunks)}块")
# 运行测试
test_text = "测试文本。" * 10000
benchmark_splitting(test_text)
实用技巧
# 1. 自动检测分割大小
def auto_detect_chunk_size(text):
"""根据文本特性自动选择合适的分块大小"""
avg_sentence_length = sum(len(s) for s in re.split(r'[。！？.!?]', text)) / max(len(re.split(r'[。！？.!?]', text)), 1)
# 推荐每块包含5-10个句子
min_chunk = max(avg_sentence_length * 5, 200)
max_chunk = min(avg_sentence_length * 10, 2000)
return int(min(max_chunk, max(min_chunk, 1000)))
# 2. 保存分割结果
def save_chunks(chunks, prefix='chunk', output_dir='./chunks'):
"""保存分割后的文本到文件"""
import os
os.makedirs(output_dir, exist_ok=True)
for i, chunk in enumerate(chunks):
filename = os.path.join(output_dir, f"{prefix}_{i+1}.txt")
with open(filename, 'w', encoding='utf-8') as f:
f.write(chunk)
print(f"已保存 {len(chunks)} 个文件到 {output_dir}")
# 使用示例
text_splitter = TextSplitter(method='sentence')
chunks = text_splitter.split(large_text)
save_chunks(chunks, 'split_text')
这些方法涵盖了大多数文本分割场景,你可以根据具体需求选择合适的方法。
