☰
文件操作:读写文件,Python是认真的
2026/9/26 4:17:52 网站建设 项目流程

该程序并不会对文件进行读写操作, 在本质上, 它仅仅被视作一个具备封闭性质的计算工具。

不管是搞数据分析、去处理日志、忙配置管理, 还是一些人愿意去写爬虫, 做文件这一套操作起来的手段, 始终得说是必备的那个本事。

这个文件的处理操作在行业内是出了名的简单和雅致, 今天我们将进行一次全面且深入的解析。

一、把文件打开, 再把文件关闭, 这就是第一点一里面讲的基本操作内容。

# 打开文件(默认是只读模式)

file = open("data.txt", "r", encoding="utf-8")

content = file.read()

file.close() # 记得关闭!否则可能丢失数据

1.第两个, 是去选择使用那个叫做with的语言语法结构, 这个方式是被大家所倡导和推荐的。

永远采用使用与的方式打开文件, 这样做它可以自动地关闭文件, 即便是发生了异常。

# 推荐写法

with open("data.txt", "r", encoding="utf-8") as file:

content = file.read()

# 到这里文件自动关闭

# 等价于:

try:

file = open("data.txt", "r", encoding="utf-8")

content = file.read()

finally:

file.close() # 保证一定会关闭

使用 with 语法所带来的好处, 是完全可以不用再手动地输入 close() 这个方法调用, 并且也就根本不会存在因为出现异常状况而忘记了关闭文件这样一个情况。

二、所谓文件模式是指该文件的操作方式。

二进制模式:加 b,如 "rb"、"wb",用于图片、音频等二进制文件。

三、在3.1节中, 我们将讨论文件读取的三种不同方式。

# 假设 data.txt 内容:

# 第一行

# 第二行

# 第三行

with open("data.txt", "r", encoding="utf-8") as f:

# 方法1:全部读取

content = f.read()

print(content) # 整个文件内容

# 方法2:逐行读取(推荐大文件)

with open("data.txt", "r", encoding="utf-8") as f:

for line in f: # 最Pythonic的方式

print(line.strip()) # strip() 去掉末尾的换行符

# 方法3:读取所有行到列表

with open("data.txt", "r", encoding="utf-8") as f:

lines = f.readlines()

print(lines) # ['第一行\n', '第二行\n', '第三行\n']

3.针对二, 大文件的处理。

# 处理1GB的文件,不能一次性读入内存

with open("big_file.log", "r", encoding="utf-8") as f:

for line in f:

# 逐行处理,内存占用很小

process(line)

# 分块读取(二进制文件)

with open("large_binary.bin", "rb") as f:

while chunk := f.read(8192): # 每次读8KB

process(chunk)

3.3 seek() 和 tell()

with open("data.txt", "r", encoding="utf-8") as f:

print(f.tell()) # 0(当前位置:字节偏移)

content = f.read(5) # 读5个字符

print(content) # 看具体情况

print(f.tell()) # 当前位置

f.seek(0) # 回到文件开头

f.seek(10) # 跳到第10个字节

f.seek(0, 2) # 跳到文件末尾(常用于获取文件大小)

四、写入文件4.1 基本写入

# 覆盖写入

with open("output.txt", "w", encoding="utf-8") as f:

f.write("第一行\n")

f.write("第二行\n")

# 追加写入

with open("output.txt", "a", encoding="utf-8") as f:

f.write("第三行(追加的)\n")

4.当执行数字二的操作, 也就是写入多行数据的时候。

lines = ["苹果\n", "香蕉\n", "橘子\n"]

with open("fruits.txt", "w", encoding="utf-8") as f:

# writelines 不会自动加换行符!

f.writelines(lines)

# 等价于:

with open("fruits.txt", "w", encoding="utf-8") as f:

for line in lines:

f.write(line)

4.3 print() 写入文件

# 利用 print 的 file 参数直接写入

with open("output.txt", "w", encoding="utf-8") as f:

print("Hello, World!

", file=f)

print("这是第二行", file=f)

print("默认会有换行", file=f)

print 这个函数在输出的时候会自动给你加一个换行符\n, 这种机制非常有助于程序员进行快速调试, 并且也很适合把相关内容直接输出到文件里面去。

五、关于路径与目录操作这一部分的内容, 在第五点一分节里, 主要讲的是os.path方法跟其他方案之间的对比情况。

import os

from pathlib import Path

# 旧方式:os.path

data_dir = os.path.join("data", "2026", "06")

print(data_dir) # data\2026\06 (Windows)

file_path = os.path.abspath("data.txt")

print(os.path.dirname(file_path)) # 目录部分

print(os.path.basename(file_path)) # 文件名

# 新方式:pathlib(Python 3.4+,推荐)

data_dir = Path("data") / "2026" / "06" # 用 / 拼接!

print(data_dir) # data\2026\06

file_path = Path("data.txt").resolve()

print(file_path.parent) # 上级目录

print(file_path.name) # data.txt

print(file_path.stem) # data(无后缀)

print(file_path.suffix) # .txt(后缀)

5.2 文件和目录的增删改查

import os

from pathlib import Path

# 检查是否存在

print(os.path.exists("data.txt")) # True/False

print(Path("data.txt").exists())

# 判断是文件还是目录

print(os.path.isfile("data.txt"))

print(os.path.isdir("data"))

# 创建目录

os.makedirs("data/subdir", exist_ok=True) # 递归创建目录

Path("data/subdir").mkdir(parents=True, exist_ok=True)

# 列出目录内容

for item in os.listdir("."):

print(item)

# 使用 pathlib 更强大

for item in Path(".").iterdir():

if item.is_file():

print(f" {item.name}")

elif item.is_dir():

print(f" {item.name}")

# glob 通配符匹配

for py_file in Path(".").glob("*.py"):

print(f"Python文件:{py_file}")

# 递归搜索

for all_py in Path(".").rglob("*.py"):

print(f"找到:{all_py}")

# 删除

os.remove("old_file.txt") # 删除文件

os.rmdir("empty_dir") # 删除空目录

import shutil

shutil.rmtree("non_empty_dir") # 递归删除(危险!

)

5.3 临时文件

import tempfile

# 创建临时文件(使用完自动删除)

with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=True) as f:

f.write("临时数据")

print(f.name) # C:\Users\...\tmpXXXX.txt

# 文件用完后自动删除

# 创建临时目录

with tempfile.TemporaryDirectory() as tmpdir:

tmp_path = Path(tmpdir) / "data.txt"

tmp_path.write_text("临时数据", encoding="utf-8")

# 目录用完后自动删除

六、CSV 文件处理

import csv

# 写入 CSV

data = [

["姓名", "年龄", "城市"],

["小明", 18, "北京"],

["小红", 20, "上海"],

["小刚", 22, "广州"],

]

with open("students.csv", "w", newline="", encoding="utf-8-sig") as f:

# utf-8-sig 可以解决 Excel 打开中文乱码

writer = csv.writer(f)

writer.writerows(data)

# 读取 CSV

with open("students.csv", "r", encoding="utf-8-sig") as f:

reader = csv.reader(f)

for row in reader:

print(f"{row[0]} | {row[1]}岁 | {row[2]}")

# 使用 DictReader(推荐,可读性更好)

with open("students.csv", "r", encoding="utf-8-sig") as f:

reader = csv.DictReader(f)

for row in reader:

print(f"{row['姓名']} | {row['年龄']}岁 | {row['城市']}")

七、JSON 文件处理

import json

# 写入 JSON

data = {

"name": "小明",

"age": 18,

"scores": [85, 92, 78],

"address": {

"city": "北京",

"district": "海淀区"

},

"is_active": True

}

with open("user.json", "w", encoding="utf-8") as f:

json.dump(data, f, ensure_ascii=False, indent=2)

# 读取 JSON

with open("user.json", "r", encoding="utf-8") as f:

loaded_data = json.load(f)

print(loaded_data["name"]) # 小明

# 快速读写(Python 3.4+?

其实可以自定义,但这里用标准方法)

# 一行读取

content = json.loads(Path("user.json").read_text(encoding="utf-8"))

八、实际操作方面, 我们会先对日志数据进行具体的分析处理, 并且会使用专门的工具来对文件进行切割分割。

# file_utils.py

from pathlib import Path

from collections import Counter

import json

import csv

from datetime import datetime

class FileAnalyzer:

"""文件分析工具箱"""

@staticmethod

def count_lines(filepath, encoding="utf-8"):

"""统计文件行数"""

with open(filepath, "r", encoding=encoding) as f:

return sum(1 for line in f)

@staticmethod

def split_file(filepath, lines_per_file=1000, encoding="utf-8"):

"""将大文件分割成多个小文件"""

filepath = Path(filepath)

base = filepath.stem

suffix = filepath.suffix

part_num = 1

current_lines = []

with open(filepath, "r", encoding=encoding) as f:

for line in f:

current_lines.append(line)

if len(current_lines) >= lines_per_file:

output_path = filepath.parent / f"{base}_part{part_num:03d}{suffix}"

with open(output_path, "w", encoding=encoding) as out:

out.writelines(current_lines)

print(f"已创建:{output_path.name} ({len(current_lines)}行)")

current_lines = []

part_num += 1

# 剩余行

if current_lines:

output_path = filepath.parent / f"{base}_part{part_num:03d}{suffix}"

with open(output_path, "w", encoding=encoding) as out:

out.writelines(current_lines)

print(f"已创建:{output_path.name} ({len(current_lines)}行)")

print(f"\n共分割成 {part_num} 个文件")

@staticmethod

def merge_files(input_dir, output_file, pattern="*", encoding="utf-8"):

"""合并多个文件"""

input_path = Path(input_dir)

files = sorted(input_path.glob(pattern))

total_lines = 0

with open(output_file, "w", encoding=encoding) as out:

for filepath in files:

with open(filepath, "r", encoding=encoding) as f:

content = f.read()

out.write(content)

total_lines += content.count("\n") + 1

print(f"已合并:{filepath.name}")

print(f"\n合并完成:{len(files)}个文件 → {output_file}")

print(f"总行数:{total_lines}")

@staticmethod

def file_report(directory="."):

"""生成目录文件报告(JSON格式)"""

dir_path = Path(directory)

report = {

"directory": str(dir_path.resolve()),

"generated_at": datetime.now().isoformat(),

"files": []

}

for item in dir_path.iterdir():

if item.is_file():

info = {

"name": item.name,

"size": item.stat().st_size,

"suffix": item.suffix,

"modified": datetime.fromtimestamp(

item.stat().st_mtime

).isoformat()

}

report["files"].append(info)

# 统计文件类型

type_count = Counter(item["suffix"] for item in report["files"])

report["file_type_summary"] = dict(type_count)

report["total_files"] = len(report["files"])

report["total_size"] = sum(item["size"] for item in report["files"])

# 保存报告

report_path = dir_path / "file_report.json"

with open(report_path, "w", encoding="utf-8") as f:

json.dump(report, f, ensure_ascii=False, indent=2)

print(f" 报告已保存:{report_path}")

print(f" 文件数:{report['total_files']}")

print(f" 总大小:{report['total_size'] / 1024:.1f} KB")

return report

# === 使用示例 ===

if __name__ == "__main__":

analyzer = FileAnalyzer()

# 1. 创建示例大文件用于测试

with open("sample_big.log", "w", encoding="utf-8") as f:

for i in range(2500):

f.write(f"第{i+1}行:这是测试数据,用于演示文件分割\n")

# 2. 分割文件

print("=== 文件分割 ===")

analyzer.split_file("sample_big.log", lines_per_file=1000)

# 3. 合并文件

print("\n=== 文件合并 ===")

analyzer.merge_files(".", "merged_output.log", pattern="sample_big_part*")

# 4. 生成报告

print("\n=== 文件报告 ===")

analyzer.file_report(".")

总结

对于文件的读写操作, 其最为核心的处理套路如下:

with open("file", "mode", encoding="utf-8") as f:

# 读或写操作

pass # 自动关闭

选对模式:

我们平时经常用到的一些帮助性的工具。

接下来将要发布后续的预告片内容, 该部分的主题聚焦于异常处理技术。我们将深入解析这套让程序运行平稳、避免意外停止且不被强制终止的关键秘密诀窍。

文件操作属于后端开发的基础内容, 大家可以收藏这篇文字, 当需要编写具体代码时再进行翻阅查看。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询