Python文本分割案例如何拆分大文本

wen python案例 2026-07-01 219

本文目录导读：

Python文本分割案例如何拆分大文本

方法1：按固定大小分割
方法2：按行分割（文件处理）
方法3：按段落分割
方法4：智能分割（保留句子完整）
方法5：使用生成器（内存友好）
实际应用示例
性能优化建议
实用技巧

我来介绍几种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')

这些方法涵盖了大多数文本分割场景,你可以根据具体需求选择合适的方法。

本文地址： https://www.bmjup.com/post/13034.html