最近在技术社区里，我注意到一个有趣的现象：很多开发者，尤其是刚接触数据处理和内容管理的同学，在处理文本文件时，常常会陷入一种“手动地狱”。比如，你从不同渠道收集了一批文本文件（TXT），它们格式混乱、编码不一、命名随意，你需要快速提取关键信息、统一格式、甚至进行简单的分析和聚合。手动打开每个文件复制粘贴？效率太低，而且容易出错。

这让我想起了一个更广泛的开发痛点： 我们如何系统化、自动化地处理非结构化的文本数据集合？ 这不是一个简单的文件合并问题，它涉及到编码识别、内容清洗、模式匹配、批量操作等一系列工程化实践。今天，我们就以处理一个名为“【TXT|崔然竣】‘黑厂牌’ 初聚会（中字）”的虚构文件集合为例，深入探讨一套完整的、可复用的本地化文本数据处理流水线。我们将完全使用 Python 标准库和主流开源工具，不依赖任何在线服务，确保流程的隐私性和可控性。

读完本文，你将能掌握：

  1. 如何构建一个健壮的脚本，自动遍历、读取并统一处理大量杂乱的 TXT 文件。
  2. 如何解决中文文本处理中最棘手的编码问题。
  3. 如何利用正则表达式和简单自然语言处理技术提取结构化信息。
  4. 如何将处理结果输出为整洁的 CSV、Markdown 或新的文本文件，便于后续分析或发布。
  5. 一套能直接应用于日志分析、数据清洗、内容归档等真实场景的工程化代码模板。

1. 这篇文章真正要解决的问题：从混乱文本到结构化数据

我们面对的往往不是单个文件，而是一个文件夹，里面散落着几十甚至上百个类似 【TXT|崔然竣】“黑厂牌” 初聚会（中字）.txt 、 崔然竣-采访片段.txt 、 black_label_meeting_notes.txt 这样的文件。文件名本身可能包含元信息（如人物、主题、语言），文件内容更是五花八门。

手动处理的成本极高，且无法规模化。真正的需求是： 自动化地完成“收集-解析-清洗-输出”的全流程 。这个过程的核心挑战不在于算法多么高深，而在于工程的鲁棒性——你的脚本要能从容应对文件缺失、编码异常、内容格式突变等情况，而不是跑一半崩溃。

本文将构建一个名为 text_pipeline.py 的处理引擎，它不只是一个脚本，更是一个可配置、可扩展的框架思路。我们会先讲清楚核心原理，然后给出每一步的代码实现，最后讨论如何将它适配到你自己的项目中。

2. 基础概念与核心原理

在开始编码前，需要明确几个关键概念，这能帮助我们在后面做出正确的技术选择。

2.1 字符编码：中文文本的“头号杀手” 为什么文本文件打开有时是乱码？根源在于编码。常见的编码有：

  • UTF-8 ：现代应用和Web的标准，兼容ASCII，是处理多语言（包括中文）的首选。
  • GBK / GB2312 ：中文Windows系统的传统默认编码。
  • ANSI ：在中文环境下通常等同于 GBK 。

原理 ：当读取文件的程序使用的编码与文件实际保存的编码不一致时，就会产生乱码。我们的处理程序必须能自动检测或尝试多种编码。

2.2 正则表达式：文本模式的“瑞士军刀” 正则表达式（Regex）是一种用于匹配字符串中字符组合的模式。在文本处理中，它不可或缺。

  • 用途 ：查找特定模式（如日期 2023-01-01 ）、提取信息（如括号内的内容）、替换文本。
  • 示例 ：从文件名 【TXT|崔然竣】“黑厂牌” 初聚会（中字）.txt 中提取 崔然竣 和 黑厂牌 。

2.3 批处理与结构化输出

  • 批处理 ：通过循环和文件系统操作，自动化处理整个目录下的文件。
  • 结构化输出 ：将提取出的信息（如文件名、人物、主题、内容摘要、字数）组织成表格形式（如CSV、JSON），便于用Excel、Pandas或数据库进行下一步分析。

下表对比了手动处理与自动化流水线的差异：

处理环节 手动处理 本文的自动化流水线
文件收集 资源管理器逐个点击 os.walk 或 glob 递归遍历
编码识别 用记事本尝试“另存为”猜测 chardet 库自动检测，多重编码回退机制
信息提取 肉眼查找、复制粘贴 预定义正则表达式模式，批量匹配提取
内容清洗 查找替换功能 使用 str.replace 或 re.sub 进行规则化清洗
结果汇总 手工复制到Excel 自动生成CSV、Markdown报告
可重复性 几乎为零，每次都是新劳动 一次编写，多次运行，结果一致
处理规模 适用于少量文件 轻松处理成千上万文件

3. 环境准备与前置条件

我们将使用 Python 作为实现语言，因为它拥有极其丰富的文本处理库和简洁的语法。请确保你的环境满足以下条件：

  • 操作系统 ：Windows 10/11, macOS, 或 Linux (如 Ubuntu)。本文代码是跨平台的。
  • Python 版本 ： Python 3.7 或更高版本 。这是许多现代库的基础要求。在终端输入 python --version 或 python3 --version 检查。
  • 包管理工具 ： pip 。通常随 Python 安装。
  • 推荐 IDE ：VS Code (配合 Python 插件) 或 PyCharm，能极大提升开发效率。

我们需要安装一个关键的第三方库来处理棘手的编码问题：

pip install chardet

chardet 库能够以较高的准确性猜测文件的字符编码。

其他我们将用到的库（ os , re , csv , json , sys ）都是 Python 标准库，无需额外安装。

创建一个新的项目文件夹，例如 text_processing_project ，并在其中开始我们的工作。

4. 核心流程拆解

我们的自动化流水线将遵循一个清晰的、模块化的流程，如下图所示（我们用文字描述这个逻辑流程）：

  1. 启动与配置 ：指定要处理的根目录，定义输出格式和提取规则。
  2. 文件遍历 ：递归地找到目标目录下所有的 .txt 文件。
  3. 文件读取与解码 ：对每个文件，尝试检测并以其正确的编码打开，读取内容。
  4. 信息提取 ：从文件名和文件内容中，应用规则提取我们关心的元数据和关键内容。
  5. 内容清洗与处理 ：对读取的文本进行清理（如去除多余空行、统一标点）。
  6. 结果聚合 ：将每个文件处理后的信息收集到一个结构化的列表（List of Dict）中。
  7. 输出与导出 ：将聚合的结果写入到 CSV、Markdown 或 JSON 文件。
  8. 日志与错误处理 ：在整个过程中记录状态，优雅地处理异常文件，不让单个错误导致整个流程中断。

这个流程的核心思想是 “管道化” ，每个环节职责单一，易于调试和扩展。

5. 完整示例与代码实现

下面，我们一步步实现这个流水线。我们将创建两个主要文件： config.py （存放配置）和 text_pipeline.py （主逻辑）。

5.1 定义配置 (config.py)

首先，我们将可配置的部分分离出来，这样以后要修改规则或输出格式时，只需改动这个文件。

# config.py
import os

# 1. 输入输出路径配置
INPUT_DIRECTORY = "./sample_texts"  # 存放你的原始TXT文件的文件夹
OUTPUT_CSV = "./output/processed_results.csv"  # CSV输出路径
OUTPUT_MD = "./output/summary.md"  # Markdown报告输出路径

# 确保输出目录存在
os.makedirs(os.path.dirname(OUTPUT_CSV), exist_ok=True)
os.makedirs(os.path.dirname(OUTPUT_MD), exist_ok=True)

# 2. 文件名解析规则 (使用正则表达式)
# 示例：从类似 `【TXT|崔然竣】“黑厂牌” 初聚会（中字）.txt` 的文件名中提取信息
FILENAME_PATTERNS = [
    {
        'name': 'brackets_pattern',
        'pattern': r'【(.*?)\|(.*?)】(.*?)（(.*?)）', # 匹配 【类型|人物】主题（语言）
        'groups': ['type', 'person', 'topic', 'language']
    },
    {
        'name': 'dash_pattern',
        'pattern': r'(.*?)-(.*?)\.txt', # 匹配 人物-主题.txt
        'groups': ['person', 'topic']
    },
    # 可以继续添加更多匹配模式
]

# 3. 内容提取规则 (同样使用正则表达式)
CONTENT_PATTERNS = {
    'date': r'\d{4}[-年]\d{1,2}[-月]\d{1,2}[日]?',  # 匹配日期，如 2023-01-01 或 2023年1月1日
    'mention_black_label': r'黑厂牌|Black Label',  # 匹配提及特定关键词
    # 可添加更多内容匹配规则
}

# 4. 文本清洗规则
CLEANING_RULES = [
    (r'\r\n', '\n'),  # 统一换行符为 \n
    (r'\n{3,}', '\n\n'),  # 将连续3个及以上换行替换为2个
    (r'[ \t]+', ' '),  # 将多个空格或制表符替换为一个空格
]

# 5. 编码检测备选列表 (按优先级排序)
ENCODING_PRIORITY = ['utf-8', 'gbk', 'gb2312', 'ascii']

5.2 实现主处理流水线 (text_pipeline.py)

这是核心文件，包含了所有的处理逻辑。

# text_pipeline.py
import os
import re
import csv
import json
import chardet
from typing import List, Dict, Any, Optional
import config  # 导入我们的配置

class TextFileProcessor:
    """文本文件处理核心类"""

    def __init__(self):
        self.results = []  # 存储所有文件处理结果的列表

    def detect_encoding(self, file_path: str) -> Optional[str]:
        """检测文件编码，如果失败则回退到配置的优先级列表"""
        try:
            with open(file_path, 'rb') as f:
                raw_data = f.read(10000)  # 读取前10000字节来检测，通常足够
                result = chardet.detect(raw_data)
                encoding = result['encoding']
                confidence = result['confidence']
                # 如果置信度太低，则回退
                if encoding and confidence > 0.7:
                    return encoding.lower()
        except Exception as e:
            print(f"  编码检测失败 {file_path}: {e}")
        # 回退到手动尝试
        for enc in config.ENCODING_PRIORITY:
            try:
                with open(file_path, 'r', encoding=enc) as f:
                    f.read(1024)  # 尝试读取一小段
                return enc
            except (UnicodeDecodeError, LookupError):
                continue
        return None  # 所有尝试都失败

    def read_file_safely(self, file_path: str) -> Optional[str]:
        """安全地读取文件，处理编码问题"""
        encoding = self.detect_encoding(file_path)
        if not encoding:
            print(f"  警告：无法确定文件编码，跳过 {file_path}")
            return None
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                content = f.read()
            return content
        except Exception as e:
            print(f"  读取文件失败 {file_path} (编码: {encoding}): {e}")
            return None

    def parse_filename(self, filename: str) -> Dict[str, str]:
        """根据配置的规则解析文件名，提取元数据"""
        meta = {'source_file': filename}
        for pattern_info in config.FILENAME_PATTERNS:
            match = re.search(pattern_info['pattern'], filename)
            if match:
                for i, group_name in enumerate(pattern_info['groups'], start=1):
                    if i <= len(match.groups()):
                        meta[group_name] = match.group(i)
                break  # 匹配到第一个规则后就退出
        return meta

    def clean_content(self, content: str) -> str:
        """应用清洗规则清理文本内容"""
        cleaned = content
        for pattern, replacement in config.CLEANING_RULES:
            cleaned = re.sub(pattern, replacement, cleaned)
        return cleaned.strip()

    def extract_from_content(self, content: str) -> Dict[str, Any]:
        """从文本内容中提取特定信息"""
        extracts = {}
        for key, pattern in config.CONTENT_PATTERNS.items():
            matches = re.findall(pattern, content)
            if matches:
                extracts[key] = matches if len(matches) > 1 else matches[0]
        # 计算一些基本统计信息
        extracts['char_count'] = len(content)
        extracts['line_count'] = content.count('\n') + 1
        # 简单的内容预览 (取前100字符)
        extracts['preview'] = (content[:97] + '...') if len(content) > 100 else content
        return extracts

    def process_single_file(self, file_path: str) -> Optional[Dict[str, Any]]:
        """处理单个文件的完整流程"""
        print(f"处理: {os.path.basename(file_path)}")
        # 1. 读取
        content = self.read_file_safely(file_path)
        if content is None:
            return None
        # 2. 解析文件名
        file_meta = self.parse_filename(os.path.basename(file_path))
        # 3. 清洗内容
        cleaned_content = self.clean_content(content)
        # 4. 内容提取
        content_extracts = self.extract_from_content(cleaned_content)
        # 5. 合并结果
        result = {
            **file_meta,  # 文件名解析出的元数据
            **content_extracts,  # 内容提取出的信息
            'cleaned_content': cleaned_content  # 可选：保存清洗后的全文
        }
        return result

    def walk_and_process(self, root_dir: str):
        """遍历目录并处理所有txt文件"""
        for dirpath, dirnames, filenames in os.walk(root_dir):
            for filename in filenames:
                if filename.lower().endswith('.txt'):
                    full_path = os.path.join(dirpath, filename)
                    result = self.process_single_file(full_path)
                    if result:
                        self.results.append(result)
        print(f"\n处理完成。共成功处理 {len(self.results)} 个文件。")

    def export_to_csv(self, output_path: str):
        """将结果导出为CSV文件"""
        if not self.results:
            print("没有结果可导出。")
            return
        # 动态获取所有可能的字段
        fieldnames = set()
        for item in self.results:
            fieldnames.update(item.keys())
        fieldnames = sorted(fieldnames)  # 排序以便阅读
        try:
            with open(output_path, 'w', newline='', encoding='utf-8-sig') as csvfile: # utf-8-sig 支持Excel中文
                writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
                writer.writeheader()
                writer.writerows(self.results)
            print(f"CSV 文件已导出至: {output_path}")
        except Exception as e:
            print(f"导出CSV失败: {e}")

    def export_to_markdown(self, output_path: str):
        """将结果导出为Markdown报告"""
        if not self.results:
            print("没有结果可导出。")
            return
        try:
            with open(output_path, 'w', encoding='utf-8') as mdfile:
                mdfile.write("# 文本处理结果汇总报告\n\n")
                mdfile.write(f"共处理文件 **{len(self.results)}** 个。\n\n")
                for i, item in enumerate(self.results, 1):
                    mdfile.write(f"## {i}. {item.get('source_file', 'N/A')}\n\n")
                    mdfile.write(f"**人物**: {item.get('person', '未提取')}  \n")
                    mdfile.write(f"**主题**: {item.get('topic', '未提取')}  \n")
                    mdfile.write(f"**字数**: {item.get('char_count', 'N/A')}  \n")
                    mdfile.write(f"**提及关键词**: {item.get('mention_black_label', '无')}  \n")
                    mdfile.write(f"**内容预览**: {item.get('preview', '')}\n\n")
                    mdfile.write("---\n\n")
            print(f"Markdown 报告已导出至: {output_path}")
        except Exception as e:
            print(f"导出Markdown失败: {e}")

def main():
    """主函数"""
    processor = TextFileProcessor()
    print(f"开始处理目录: {config.INPUT_DIRECTORY}")
    processor.walk_and_process(config.INPUT_DIRECTORY)
    processor.export_to_csv(config.OUTPUT_CSV)
    processor.export_to_markdown(config.OUTPUT_MD)
    print("所有任务执行完毕。")

if __name__ == "__main__":
    main()

5.3 创建示例数据并运行

为了测试，我们在项目根目录下创建一个 sample_texts 文件夹，并放入几个测试文件。

文件 1: 【TXT|崔然竣】“黑厂牌” 初聚会（中字）.txt 

日期：2023-10-26
地点：首尔某工作室

本次“黑厂牌”初聚会由崔然竣发起，主要讨论了团队未来的音乐方向。
成员们就“Black Label”这个品牌概念进行了深入交流。
会议持续了约3小时，气氛热烈。

关键词：黑厂牌， 音乐制作， 团队协作。

文件 2: 崔然竣-个人访谈片段.txt 

采访者：可以谈谈你对“黑厂牌”的理解吗？
崔然竣：我认为“黑厂牌”不仅仅是一个名字，它代表了一种追求极致和创新的精神。
我们希望做出有辨识度的音乐。

（采访时间：2023年11月）

文件 3: notes_black_label_project.txt (这是一个用GBK编码保存的文件，用于测试编码自动检测)

这是一个关于黑厂牌项目的简单备忘录。
需要协调录音室时间。
下一步：联系制作人。

现在，在终端运行我们的主脚本：

cd /path/to/your/text_processing_project
python text_pipeline.py

6. 运行结果与效果验证

运行脚本后，你将在控制台看到类似输出：

开始处理目录: ./sample_texts
处理: 【TXT|崔然竣】“黑厂牌” 初聚会（中字）.txt
处理: 崔然竣-个人访谈片段.txt
处理: notes_black_label_project.txt

处理完成。共成功处理 3 个文件。
CSV 文件已导出至: ./output/processed_results.csv
Markdown 报告已导出至: ./output/summary.md
所有任务执行完毕。

验证输出 1：CSV 文件 用 Excel 或文本编辑器打开 ./output/processed_results.csv ，你会看到一个结构清晰的表格，包含以下列（字段名可能因匹配结果略有不同）：

source_file type person topic language date mention_black_label char_count line_count preview
【TXT 崔然竣】“黑厂牌” 初聚会（中字）.txt TXT 崔然竣 “黑厂牌” 初聚会 中字 2023-10-26 黑厂牌 ... ...
崔然竣-个人访谈片段.txt 崔然竣 个人访谈片段 2023-11 黑厂牌 ... ... 采访者：可以谈谈你对“黑厂牌”的理解吗？...
notes_black_label_project.txt 黑厂牌 ... ... 这是一个关于黑厂牌项目的简单备忘录...

这个 CSV 文件可以直接导入到数据库或使用 Pandas 进行进一步分析。

验证输出 2：Markdown 报告 打开 ./output/summary.md ，你会得到一份可读性很强的报告，概述了每个文件的关键信息，非常适合快速浏览或分享。

如何判断成功？

  1. 控制台无报错 ：除了可能的“警告”信息，没有出现红色的异常堆栈跟踪。
  2. 输出文件生成 ：在 ./output/ 目录下成功生成了两个文件。
  3. 内容正确提取 ：打开生成的文件，检查文件名中的“人物”、“主题”是否被正确解析，内容中的“日期”、“关键词”是否被提取。
  4. 编码处理正常 ：第三个测试文件 ( notes_black_label_project.txt ) 即使使用 GBK 编码，内容也应被正确读取，没有乱码。

如果运行失败，第一步应检查：

  • Python 版本是否为 3.7+。
  • 是否在项目根目录下运行。
  • sample_texts 文件夹和测试文件是否存在。
  • 是否已安装 chardet 库 ( pip install chardet )。

7. 常见问题与排查思路

在实际使用中，你可能会遇到以下问题。这里提供系统的排查思路。

问题现象 可能原因 排查方式 解决方案
运行脚本后无任何输出，立即退出 1. 脚本存在语法错误。
2. if __name__ == "__main__": 前的代码有错误。		 在终端使用 python -m py_compile text_pipeline.py 检查语法。或在脚本开头加 print(“开始”) 测试。 根据错误信息修正语法。确保代码缩进正确。
提示 ModuleNotFoundError: No module named 'chardet' 未安装 chardet 库。 在终端运行 pip list 查看已安装包。 运行 pip install chardet 安装。
部分文件内容读取为乱码 1. 文件编码非常见。
2. 文件本身已损坏或混合编码。		 1. 检查 config.ENCODING_PRIORITY 列表是否包含该编码（如 gb18030 , big5 ）。
2. 用二进制模式 ( rb ) 打开文件，查看其头部字节。		 1. 将可疑编码添加到 ENCODING_PRIORITY 列表前端。
2. 对于复杂情况，可尝试使用 codecs 库的 open 函数并指定 errors='ignore' 或 errors='replace' 。
文件名解析规则未匹配，元数据为空 实际文件名格式与 config.FILENAME_PATTERNS 中定义的正则表达式不匹配。 1. 打印出实际的文件名。
2. 使用在线正则测试工具（如 regex101.com）调试你的正则表达式。		 1. 在 config.py 中添加或修改 FILENAME_PATTERNS 以匹配你的文件名格式。
2. 确保正则表达式中的分组 () 与 groups 列表对应。
内容提取规则未匹配到预期关键词 1. 正则表达式写错。
2. 文本中存在全角/半角符号差异或空格。		 1. 将 config.CONTENT_PATTERNS 中的模式复制到正则测试工具中，用实际文本测试。
2. 打印出 cleaned_content 的前几百字符，检查文本是否已被正确清洗。		 1. 修正正则表达式，例如使用 r'黑\s*厂\s*牌' 来匹配可能被空格隔开的情况。
2. 在清洗规则中增加对全角字符的转换或更灵活的空格处理。
处理大量文件时内存占用过高或速度慢 1. 一次性将所有文件内容读入内存的列表 ( self.results )。
2. 未对超大文件进行分块读取。		 监控任务管理器的内存使用情况。 1. 对于海量文件，考虑流式处理：处理完一个文件后立即写入输出流（如CSV writer），而不是累积在内存中。
2. 对于单个超大文件，在 read_file_safely 中可以考虑分块读取和处理。
生成的CSV文件用Excel打开中文乱码 Excel 默认可能不识别不带 BOM 的 UTF-8 编码。 用纯文本编辑器（如VS Code, Notepad++）打开CSV，看中文是否正常。 在 export_to_csv 函数中，将 encoding='utf-8' 改为 encoding='utf-8-sig' ，这会添加BOM头，Excel即可正确识别。

8. 最佳实践与工程建议

将脚本提升到“工程化”水平，才能在生产环境中可靠运行。

8.1 配置与代码分离 正如我们所做的，将所有可变的规则、路径、参数放在 config.py 中。这样，业务逻辑 ( text_pipeline.py ) 保持稳定，而需求变化时只需修改配置。你甚至可以支持从 JSON 或 YAML 文件加载配置。

8.2 完善的日志记录 当前的 print 语句不利于长期维护。应引入 logging 模块，区分 DEBUG 、 INFO 、 WARNING 、 ERROR 级别，并可以输出到文件。

import logging
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s',
                    handlers=[logging.FileHandler('pipeline.log'), logging.StreamHandler()])

8.3 异常处理与容错 我们的 process_single_file 方法已经返回 None 来处理单个文件的失败，避免整个流程中断。对于更关键的任务，可以设计重试机制或将失败文件记录到单独列表，供后续人工复查。

8.4 性能考量

  • 并发处理 ：如果文件数量巨大（>1000），且处理是CPU密集型（如复杂正则匹配），可以考虑使用 concurrent.futures.ThreadPoolExecutor 进行多线程处理（I/O密集型）或多进程处理（CPU密集型）。
  • 内存管理 ：如前所述，流式处理是处理海量数据的关键。

8.5 扩展性设计

  • 添加新的提取器 ：可以设计一个插件系统，通过继承一个基类来添加新的内容提取逻辑。
  • 支持更多文件格式 ：修改 walk_and_process 方法中的文件后缀判断，并针对 .docx , .pdf 等格式引入相应的解析库（如 python-docx , PyPDF2 ）。
  • 输出到数据库 ：在 export_to_csv 旁边，可以轻松添加一个 export_to_database 方法，使用 SQLAlchemy 或直接 DB-API 连接将结果写入 MySQL、PostgreSQL 或 SQLite。

8.6 版本控制与测试 将整个项目（除输入数据外）纳入 Git 版本控制。为核心函数（如 parse_filename , clean_content ）编写单元测试，确保修改规则时不会破坏原有功能。可以使用 pytest 框架。

9. 总结与后续学习方向

通过本文，我们不仅仅写了一个处理“【TXT|崔然竣】‘黑厂牌’ 初聚会（中字）”这类文件的脚本，而是构建了一个 本地化、自动化、可扩展的文本数据处理流水线原型 。它解决了从杂乱无章的原始文本文件中自动提取、清洗、结构化信息的核心工程问题。

本文的核心价值点在于：

  1. 编码问题的系统性解决 ：通过 chardet 检测和优先级回退，基本覆盖了中文环境下的编码难题。
  2. 规则与逻辑分离 ：所有解析和清洗规则都放在配置文件中，使核心代码稳定、易维护。
  3. 完整的处理闭环 ：从遍历、读取、解析、清洗到导出为多种格式，形成了一个完整解决方案。
  4. 生产级别的考量 ：包含了错误处理、日志、可扩展性设计和最佳实践建议。

你可以立即动手做的：

  1. 替换数据源 ：将 config.INPUT_DIRECTORY 指向你真实的、杂乱的项目文档、日志或笔记文件夹，运行脚本看看能提取出什么。
  2. 自定义规则 ：根据你的文件命名习惯，修改 config.FILENAME_PATTERNS 中的正则表达式。根据你想提取的内容（如邮箱、电话、特定编号），修改 config.CONTENT_PATTERNS 。
  3. 集成到工作流 ：将这个脚本设置为定时任务（如使用 cron 或 Windows 任务计划程序），定期处理某个目录下新增的文本文件。

进一步深入的方向：

  • 自然语言处理 (NLP) ：在提取信息后，可以使用 jieba （中文分词）、 snownlp （情感分析）或 spaCy （实体识别）进行更深度的内容分析，例如自动分类、情感判断、关键实体抽取。
  • 图形化界面 (GUI) ：使用 PyQt 、 Tkinter 或 Streamlit 为这个流水线包装一个简单的桌面或Web界面，让非技术同事也能使用。
  • 工作流自动化 ：将本脚本作为一环，与云存储（监听网盘新增文件）、消息通知（处理完成后发送邮件或钉钉消息）等其他系统集成。

处理本地文本数据是许多开发者和数据分析师的日常需求。掌握这套方法，意味着你拥有了将“脏数据”转化为“可用资产”的基础能力。建议收藏本文，并将代码作为你下一个数据处理项目的起点。

# 正则表达式
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

Manim 数学动画完全指南：五分钟从零到成片

Manim 是用 Python 驱动的数学动画引擎，把抽象的公式与曲线转成逐帧可控的动态可视化。适合想制作 Python 数学可视化视频的编程新手、数学学生和科普创作者。## 数学为什么需要"动"起来你试过用一张静态图讲清导数的概念吗？切线、割线、极限位置都能画出来，但"割线逼近切线"发生的那一下，观众只能靠脑补。课本插图大多是"结论截图"：只留下最终状态，过程被压缩成一句"易见"。函

avatar 码道开发者社区
cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar 码道开发者社区
cover

GLM-5.3-Flash深度解析：3200亿参数只卖Opus四十分之一，国产算力首次跑通前沿模型

avatar 码道开发者社区