TXT文件操作

Python 纯文本处理：TXT 文件读写与上下文管理器精讲

小象实战讲义 · Python数据分析实战

在日常数据处理与日志分析中，纯文本文件（TXT / Log）是最为基础且无处不在的数据形态。无论是服务器访问流水、设备传感器的原始报文，还是自然语言语料，几乎所有非结构化数据在初识阶段都以纯文本存储。掌握 Python 内置的文件 I/O 机制、字符编码控制以及现代推荐的 with 上下文管理器，是构建健壮数据清洗流水线的基石。本节将系统剖析 TXT 文件的打开模式、逐行流式读取与高效写入操作。

💡 核心导读

• 文件底层与行结束符：深入理解 EOL（\n）与字符编码（utf-8 vs gbk）的底层机理。
• 文件打开模式精析：掌握 'r'（只读）、'w'（覆盖写）与 'a'（追加写）的区别与安全准则。
• 读写核心 API 对比：辨析 read()、readline()、readlines() 以及迭代器遍历的内存表现。
• 上下文管理器 with：掌握免显式 close() 的资源释放与异常安全保证机制。
• 日志清洗与统计实战：通过 Python 代码实现原始服务器访问日志的提取、错误过滤与汇总统计。

1. 文本文件底层基础与打开模式

纯文本文件在底层本质上是由字符编码序列构成的字节流。在文本文件中，行与行之间通过行结束符（EOL, End Of Line）隔开。在 Linux/macOS 下为 \n（Line Feed），在 Windows 下传统为 \r\n（Carriage Return + Line Feed），Python 会自动进行跨平台换行符平滑适配。

1.1 文件打开与模式选择

Python 通过内置函数 open() 与操作系统文件系统进行交互：

file_obj = open(file_path, mode='r', encoding='utf-8')

• file_path：目标文件的绝对路径或相对路径；
• mode：文件打开模式；
• encoding：字符编码（在处理中文数据时强烈建议显式指定 encoding='utf-8'，避免 Windows 默认使用 GBK 导致乱码崩溃）。

2. 核心读取操作与内存策略对比

Python 提供了多种从文本文件中读取内容的方式，我们需要根据数据规模合理选择以避免内存溢出（OOM）：

┌─────────────────────────────────────────────────────────────┐
│                    Python 文本文件读取方式对比               │
├─────────────────────────────────────────────────────────────┤
│ 1. f.read()        ➔ 一次性将整个文件读入为单个大字符串 (适合小文件)  │
│ 2. f.readline()    ➔ 每次仅读取当前单行字符串 (适合精细控制)         │
│ 3. f.readlines()   ➔ 将所有行读取为一个字符串列表 (包含 \n 换行符)   │
│ 4. for line in f:  ➔ 流式迭代器逐行读取 (🔥大文件处理黄金推荐法则)  │
└─────────────────────────────────────────────────────────────┘

• 大文件逐行流式读取法则： 当处理几个 GB 的大型日志文件时，严禁使用 f.read() 或 f.readlines()，因为它们会试图将全部内容塞进内存；使用 for line in f: 语法，Python 会通过内置缓冲生成器逐行读取，内存占用始终保持在几 KB 的极低水平。

3. 上下文管理器 with：现代 Python 规范

在早期代码中，打开文件后必须手动调用 file_obj.close() 关闭文件句柄。如果中间发生异常，close() 就可能被跳过，导致操作系统文件句柄泄露或数据未落盘。

现代 Python 强制推荐使用 with 上下文管理器：

with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()
# 缩进块结束后，Python 会自动、安全地执行文件关闭与资源释放操作，即使发生异常也不例外

4. Python 代码实战：服务器访问日志的生成、流式清洗与统计

下面我们通过一段完整的 Python 代码，演示如何使用 with 语法写入模拟日志，并通过流式迭代器完成日志的清洗提取与状态码统计。

# 示例 1：使用 with 语句创建并写入原始服务器访问日志

import os

log_file_path = "server_access.log"

# 模拟要写入的原始日志行（IP、访问时间、请求路径、HTTP状态码）
initial_logs = [
    "192.168.1.10 - [22/Aug/2026:10:00:01] \"GET /index.html HTTP/1.1\" 200\n",
    "192.168.1.15 - [22/Aug/2026:10:00:05] \"POST /api/login HTTP/1.1\" 200\n",
    "192.168.1.20 - [22/Aug/2026:10:00:12] \"GET /images/logo.png HTTP/1.1\" 304\n",
    "192.168.1.33 - [22/Aug/2026:10:00:18] \"GET /hidden/admin HTTP/1.1\" 404\n",
    "192.168.1.10 - [22/Aug/2026:10:00:25] \"POST /api/checkout HTTP/1.1\" 500\n"
]

# 1. 覆盖写入初始日志
with open(log_file_path, "w", encoding="utf-8") as f:
    f.writelines(initial_logs)
print(f"✅ 成功生成初始日志文件: {log_file_path} (共 {len(initial_logs)} 行)")

# 2. 追加一条新的告警日志
with open(log_file_path, "a", encoding="utf-8") as f:
    f.write("192.168.1.99 - [22/Aug/2026:10:00:30] \"GET /api/status HTTP/1.1\" 200\n")
print("✅ 成功追加一条新日志记录！")

接下来，我们以流式生成器迭代的方式读取该日志文件，过滤出异常请求（状态码 >= 400）并统计各 IP 的访问频次：

# 示例 2：流式逐行解析日志、提取错误请求并汇总 IP 访问统计

ip_counter = {}
error_logs = []

# 使用流式迭代器逐行扫描日志文件
with open(log_file_path, "r", encoding="utf-8") as f:
    for line_no, line in enumerate(f, start=1):
        clean_line = line.strip()
        if not clean_line:
            continue
        
        # 简单拆分字段
        parts = clean_line.split()
        ip = parts[0]
        status_code = int(parts[-1])
        
        # 统计 IP 访问次数
        ip_counter[ip] = ip_counter.get(ip, 0) + 1
        
        # 筛选异常状态码 (4xx / 5xx)
        if status_code >= 400:
            error_logs.append((line_no, ip, status_code, clean_line))

print("\n=== 日志分析汇总报告 ===")
print(f"1. IP 访问频次统计:")
for ip, count in sorted(ip_counter.items(), key=lambda x: x[1], reverse=True):
    print(f"   • IP: {ip:<15} | 访问次数: {count} 次")

print(f"\n2. 发现 {len(error_logs)} 条异常错误请求:")
for line_no, ip, code, raw in error_logs:
    print(f"   [行号 {line_no}] 来源: {ip} | 状态码: {code} | 详情: {raw}")

# 清理临时测试文件
if os.path.exists(log_file_path):
    os.remove(log_file_path)

📝 动手练一练

1. 陷阱分析题：在 Windows 系统上，如果在调用 open("notes.txt", "w") 时没有显式传入 encoding="utf-8" 参数，写入包含中文字符的文本可能会引发什么潜在问题？如何彻底规避？

   👉 点击查看参考答案

   参考答案： 在 Windows 系统上，Python 默认采用系统的 ANSI 编码（中文 Windows 下通常为 GBK / CP936）。如果未指定 encoding='utf-8'，写出的文件就是 GBK 编码；当被部署在 Linux 服务器或与其他 UTF-8 系统交互读取时，就会出现严重的中文乱码（如 UnicodeDecodeError）。 规避准则：在任何涉及文件读写的地方，始终强制显式声明 open(..., encoding='utf-8')。

2. 编程练习：请编写一段 Python 脚本，将一个包含 4 位员工姓名的列表 names = ["张伟", "王芳", "李娜", "刘洋"] 逐行写入名为 staff.txt 的文本文件中；然后再用 with 语句读取该文件，并打印出所有字数为 2 个字的员工姓名。

   👉 点击查看参考答案

   参考答案：

   import os
   
   names = ["张伟", "王芳", "李娜", "刘洋"]
   file_name = "staff.txt"
   
   # 1. 逐行写入
   with open(file_name, "w", encoding="utf-8") as f:
       for name in names:
           f.write(name + "\n")
           
   # 2. 读取并筛选 2 字姓名
   print("=== 2 字员工姓名名单 ===")
   with open(file_name, "r", encoding="utf-8") as f:
       for line in f:
           clean_name = line.strip()
           if len(clean_name) == 2:
               print(f"• {clean_name}")
               
   if os.path.exists(file_name):
       os.remove(file_name)

本章小结

在本节中，我们全面掌握了 Python 纯文本 I/O 的规范做法：

• 理解文件打开的三大核心模式（'r'、'w'、'a'），始终显式指定 encoding='utf-8'；
• 大数据量文本读取必须使用 for line in f: 流式迭代，避免内存溢出；
• 始终使用 with 上下文管理器来确保资源的确定性自动释放。

📋 行动清单

• 检查个人项目代码，将所有旧式的 f = open(); f.close() 改写为更安全的 with open() 结构。
• 掌握 strip()、split() 与字符串格式化在文本数据预处理中的组合应用。

—— 小象教研组