text_processor.py

"""
text_processor.py —— 阶段2:文本清洗与智能分块（新手必读！）

【小白解释：这个文件做什么？】
这个文件负责"清洗"和"分块"两件事：
1. 清洗：把文档解析出来的原始文字，去掉多余的空格、乱码、换行等
2. 分块：把长文档切成小块，每块大小合适，方便后续处理

【小白解释：为什么要清洗？】
PDF 解析出来的文字经常有很多"噪音"：
- 多余的空格和换行
- 不可见的控制字符（看起来像乱码）
- 连续的空白行
这些噪音会影响后续的检索效果，所以要清理掉。

【小白解释：为什么要分块？】
大模型一次能处理的文字有限（叫"上下文窗口"），
太长的文档塞进去会报错或效果变差。
所以要切成小块，每块单独存进向量库。

职责:把 file_parser.py 解析出来的"原始文本",处理成适合后续
嵌入模型使用的"干净文本块"列表。

对应作业要求的"文本预处理与智能分块":
1. 文本清洗:去除多余换行、空格、特殊字符、乱码
2. 文本分块:固定长度分块+重叠补全机制,避免上下文断裂

这两件事分两个函数做:clean_text() 和 chunk_text()。
main.py 调用的时候是这样串起来的:
    raw_text = parse_document(file_path).text
    clean = clean_text(raw_text)
    chunks = chunk_text(clean)
"""

import re  # 【小白解释】re 是 Python 的"正则表达式"模块，用来匹配和替换文本模式
from typing import List  # 【小白解释】List 是类型提示，表示"这是一个列表"

import config


# ========== 第一部分:文本清洗 ==========

def clean_text(text: str) -> str:
    """
    【小白解释】清洗原始文本，去掉"噪音"
    
    清洗原始文本,去除无效内容。

    处理顺序是有讲究的,后一步往往依赖前一步处理完的结果,
    所以不要随意调整顺序:
    1. 统一换行符(Windows的\\r\\n和Mac/Linux的\\r,统一成\\n,
       不然后面用\\n做判断的逻辑会漏掉一部分)
    2. 去除不可见的控制字符(这是PDF解析常见"乱码"的来源之一,
       比如某些特殊编码的空白符、PDF内部的格式控制符)
    3. 把连续的空格/Tab合并成一个空格
    4. 把连续3个以上的空行,压缩成最多1个空行(保留段落感,但不要一大段空白)
    5. 去掉每一行首尾多余的空格,并丢弃处理完之后变成空字符串的行
    """
    if not text:
        return ""

    # 【小白解释】步骤1：统一换行符
    # Windows 用 \r\n，Mac/Linux 用 \n，这里统一成 \n
    # .replace() 是字符串替换方法，把第一个参数替换成第二个参数
    text = text.replace("\r\n", "\n").replace("\r", "\n")

    # 【小白解释】步骤2：去除控制字符
    # 控制字符是 ASCII 编码中不可打印的字符（如换页符、退格符等）
    # 正则表达式 [\x00-\x08\x0b\x0c\x0e-\x1f\x7f] 匹配这些控制字符
    # \x00 表示 ASCII 编码为 0 的字符，\x7f 表示编码为 127 的字符
    # 注意：\t（制表符，编码9）和 \n（换行符，编码10）要保留，所以正则里跳过了它们
    text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)

    # 【小白解释】步骤3：合并连续空格
    # 正则表达式 [ \t]+ 匹配"一个或多个空格或制表符"
    # 把它们全部替换成一个空格
    text = re.sub(r"[ \t]+", " ", text)

    # 【小白解释】步骤4：压缩连续换行
    # 正则表达式 \n{3,} 匹配"3个或更多换行符"
    # 把它们替换成2个换行符（也就是两段文字之间保留1个空行）
    # 解释：\n\n 表示两个换行符，中间有一个空行，段落之间有间隔
    text = re.sub(r"\n{3,}", "\n\n", text)

    # 【小白解释】步骤5：逐行清理
    # split("\n") 把文本按换行符拆成列表
    # [line.strip() for line in ...] 是列表推导式，对每一行去掉首尾空格
    lines = [line.strip() for line in text.split("\n")]
    # 【小白解释】过滤掉空行（只保留有内容的行）
    # if line 表示"如果 line 不为空字符串"
    lines = [line for line in lines if line]

    # 【小白解释】最后把所有行用换行符连接起来，返回清洗后的文本
    return "\n".join(lines).strip()


# ========== 第二部分:智能分块 ==========

# 【小白解释】句子结束的标志符号
# 分块时优先在这些符号后面切，而不是在句子中间硬切。
# 这样切出来的块，结尾大概率是一个完整句子，语义不会断。
# 中英文标点都覆盖到，因为文档里经常中英文混排。
SENTENCE_ENDINGS = "。！？.!?\n；;：:…—\"」』）)"


def chunk_text(text: str, chunk_size: int = config.CHUNK_SIZE, overlap: int = config.OVERLAP) -> List[str]:

    if not text:
        return []

    # 防御性检查：overlap 不能大于等于 chunk_size
    # 否则切块位置不会前进，会导致死循环（下一块的起点比这一块的终点还靠前）
    # 这里强制做一个合理性校正：overlap 设为 chunk_size 的四分之一
    if overlap >= chunk_size:
        overlap = chunk_size // 4

    SEARCH_WINDOW = 100  # 往前找标点符号时，最多往前看多少个字符
    chunks = []  # 用列表收集切出来的文本块
    start = 0    # 当前块的起始位置（从0开始）
    text_length = len(text)  # 文本总长度

    # while 循环：只要还没切到文本末尾，就继续切
    while start < text_length:
        # 第一步：算出"理想情况下"这一块切到哪个位置
        # ideal_end = start + chunk_size，但不能超过文本末尾
        ideal_end = min(start + chunk_size, text_length)

        if ideal_end >= text_length:
            # 已经到文本末尾了，剩下的内容全部作为最后一块，不用再找切点
            end = text_length
        else:
            # 第二步：从 ideal_end 往前找最近的句子结束符
            end = ideal_end
            # search_start 是往前搜索的起点，不能比当前块的起点还靠前
            search_start = max(start, ideal_end - SEARCH_WINDOW)
            found_cut_point = False

            for i in range(ideal_end, search_start, -1):
                # text[i-1] 是当前位置的前一个字符
                # 如果这个字符是句子结束符，就在这里切
                if text[i - 1] in SENTENCE_ENDINGS:
                    end = i  # 在标点符号后面切，把标点留在这一块里
                    found_cut_point = True
                    break  # 找到了就跳出循环

            # 没找到合适的标点，就维持 ideal_end 硬切（兜底方案）
            if not found_cut_point:
                end = ideal_end

            # 关键防御：防止死循环
            # 如果按句子切点算出来的 end，导致下一块的起点 (end - overlap) 
            # 不比当前 start 更靠前，说明这个切点离 start 太近，
            # 继续用它会让 start 原地不动甚至倒退，陷入死循环。
            # 解决办法：放弃这个"找标点"的切点，直接用固定长度硬切
            if end - overlap <= start:
                end = ideal_end

        # 切出这一块的内容：text[start:end]
        # .strip() 去掉首尾空白
        chunk = text[start:end].strip()
        if chunk:  # 只有非空的块才加入列表
            chunks.append(chunk)

        if end >= text_length:
            break  # 已经切到末尾，跳出循环

        # 下一块的起点 = 这一块的终点 往前回退 overlap 个字符
        # 这就是"重叠补全"——下一块会重新包含上一块末尾的一部分内容
        # 这样即使关键信息被切散了，至少有一块能保留完整信息
        next_start = end - overlap

        # 双重保险：不管前面的逻辑多复杂，
        # 这里始终强制 start 必须严格大于上一轮的 start。
        # 哪怕以后有人改了上面的代码引入新漏洞，这一行也能兜底，绝对不会死循环。
        start = max(next_start, start + 1)

    return chunks