文件操作——读写文件，路径处理

概念讲解

程序的数据存在内存中，断电就没了。文件让数据持久化——读写文件是程序与外界交互的重要方式。

Python 操作文件的核心模式：

with open("文件路径", "模式", encoding="utf-8") as f:
    # 读取：content = f.read()
    # 写入：f.write(content)
# with 语句会在代码块结束后自动关闭文件

文件模式：

• r：读取（默认）
• w：写入（会覆盖原文件）
• a：追加
• rb/wb：二进制模式

路径处理：

• 使用 pathlib.Path（推荐），跨平台兼容更好
• 拼接路径用 / 或 Path.joinpath()
• 绝对路径 vs 相对路径要注意

代码示例

读取CSV文件并统计

from pathlib import Path

def analyze_csv(file_path):
    """读取CSV文件，统计每列的数据"""
    records = []
    
    with open(file_path, 'r', encoding='utf-8') as f:
        header = f.readline().strip().split(',')
        for line in f:
            line = line.strip()
            if not line:
                continue
            values = line.split(',')
            records.append(dict(zip(header, values)))
    
    # 统计每列
    stats = {}
    for col in header:
        values = [r[col] for r in records]
        stats[col] = {
            'count': len(values),
            'unique': len(set(values)),
            'sample': values[:3]
        }
    return stats

# 假设有这样的CSV（sales.csv）：
# date,product,sales
# 2024-01,手机,5000
# 2024-01,电脑,8000
# 2024-02,手机,6000

# result = analyze_csv('sales.csv')
# print(result)

批量文件处理

from pathlib import Path

def batch_rename(directory, old_pattern, new_pattern):
    """
    批量重命名文件
    例: batch_rename('C:/data', '.txt', '_backup.txt')
    """
    dir_path = Path(directory)
    renamed = []
    
    for file in dir_path.iterdir():
        if file.is_file() and file.suffix == old_pattern:
            new_name = file.stem + new_pattern + file.suffix
            new_path = file.with_name(new_name)
            file.rename(new_path)
            renamed.append((file.name, new_name))
    
    return renamed

def find_large_files(directory, size_mb=10):
    """找出超过指定大小的文件"""
    dir_path = Path(directory)
    large_files = []
    
    for file in dir_path.rglob('*'):  # rglob 递归搜索
        if file.is_file():
            size = file.stat().st_size / (1024 * 1024)
            if size > size_mb:
                large_files.append((str(file), round(size, 2)))
    
    return sorted(large_files, key=lambda x: x[1], reverse=True)

# 找出D盘下大于50MB的文件
# for path, size in find_large_files('D:/', 50):
#     print(f"{size}MB - {path}")

写入数据到文件

from pathlib import Path
import json

def export_to_json(data, file_path):
    """将数据导出为JSON文件"""
    path = Path(file_path)
    path.parent.mkdir(parents=True, exist_ok=True)  # 自动创建目录
    
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

# 使用
data = {
    "students": [
        {"name": "张三", "score": 92},
        {"name": "李四", "score": 88}
    ]
}
export_to_json(data, "output/student_report.json")

练习题

题目：写一个日志分析程序，读取一个日志文件，找出所有错误（ERROR）级别的日志，按日期统计错误数量，并输出到新文件。

日志格式示例：

2024-03-15 10:23:45 INFO 用户登录成功
2024-03-15 10:24:12 ERROR 数据库连接失败
2024-03-15 10:25:01 ERROR 请求超时
2024-03-16 09:00:00 ERROR 数据库连接失败

from pathlib import Path
from collections import defaultdict

def analyze_errors(log_file, output_file="error_summary.txt"):
    error_by_date = defaultdict(list)
    
    with open(log_file, 'r', encoding='utf-8') as f:
        for line in f:
            if ' ERROR ' in line:
                # 提取日期（前10个字符）
                date = line[:10]
                error_by_date[date].append(line.strip())
    
    # 写入结果
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write("错误日志统计\n")
        f.write("=" * 40 + "\n\n")
        
        for date, errors in sorted(error_by_date.items()):
            f.write(f"{date}：共 {len(errors)} 条错误\n")
            for err in errors:
                f.write(f"  {err}\n")
            f.write("\n")
        
        total = sum(len(e) for e in error_by_date.values())
        f.write(f"总计：{total} 条错误\n")
    
    return output_file

# 使用
# analyze_errors('app.log', 'error_report.txt')