🐍 Python Day16：文件操作 — 让程序和文件对话

🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握 open()、读写模式、with 语句、read/readline/readlines

📖 今日目录

1. 为什么要学文件操作？
2. open() 与 close()
3. with 语句（推荐写法）
4. 读文件：三种方式
5. 写文件：w / a / x 模式
6. 读写模式大全
7. 文件指针与 seek
8. 实战练习
9. 今日小结

1. 为什么要学文件操作？

程序一关，内存里的数据就没了。想让数据"活下去"，必须存到文件里。

# 没有文件操作：程序结束，数据消失
scores = {"张三": 85, "李四": 92}
# 程序关闭 → scores 不见了

# 有文件操作：数据持久化
# 程序结束前 → 存入文件
# 程序启动后 → 从文件读取
# 数据永远不会丢！

2. open() 与 close()

📖 基本语法

# 创建测试文件
with open("test.txt", "w", encoding="utf-8") as f:
    f.write("Hello Python!\n")
    f.write("这是第二行\n")
    f.write("这是第三行\n")

# 打开文件（手动关闭）
f = open("test.txt", "r", encoding="utf-8")
content = f.read()
print(content)
f.close()  # ⚠️ 必须手动关闭！

⚠️ 忘了 close() 会怎样？
1. 文件资源被占用，内存泄漏
2. 写入的数据可能没刷到磁盘（数据丢失）
3. 其他程序可能无法访问这个文件

解决方案：用 with 语句，自动关闭！

3. with 语句（推荐写法）

with 是 Python 的"安全锁"——用完文件自动关闭，即使出错也不怕。

# ✅ 推荐写法：with 语句
with open("test.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)
# 离开 with 块，文件自动关闭（不用写 f.close()）

# 等价于（底层原理）：
f = open("test.txt", "r", encoding="utf-8")
try:
    content = f.read()
    print(content)
finally:
    f.close()  # 无论如何都会关闭

# 读文件的三种模式
# 1. 读取全部内容
with open("test.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)

# 2. 逐行读取（大文件推荐）
with open("test.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())

# 3. 读取为列表
with open("test.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()  # 返回列表
    print(lines)

💡 with 语句的三大好处：
1. 自动关闭文件，不用担心忘记
2. 即使代码出错（异常），也能保证关闭
3. 代码更简洁，Pythonic 写法

4. 读文件：三种方式

📖 read() — 一次读全部

with open("test.txt", "r", encoding="utf-8") as f:
    content = f.read()       # 读全部内容（字符串）
    print(content)

# read(n) — 只读前 n 个字符
with open("test.txt", "r", encoding="utf-8") as f:
    chunk = f.read(5)        # 读前5个字符
    print(chunk)             # "Hello"

📖 readline() — 一次读一行

with open("test.txt", "r", encoding="utf-8") as f:
    line1 = f.readline()     # 第一行
    line2 = f.readline()     # 第二行
    line3 = f.readline()     # 第三行
    line4 = f.readline()     # 空字符串（文件末尾）

    print(repr(line1))       # 'Hello Python!\n'
    print(repr(line2))       # '这是第二行\n'
    print(repr(line3))       # '这是第三行\n'
    print(repr(line4))       # ''（读到末尾返回空）

📖 readlines() — 全部读成列表

with open("test.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()
    print(lines)
    # ['Hello Python!\n', '这是第二行\n', '这是第三行\n']

# 去掉每行末尾的换行符
with open("test.txt", "r", encoding="utf-8") as f:
    lines = [line.strip() for line in f.readlines()]
    print(lines)
    # ['Hello Python!', '这是第二行', '这是第三行']

📋 三种读取方式对比

💡 大文件怎么读？
千万不要用 read() 读几个 GB 的文件！内存会爆。
用 for line in f 逐行迭代，每次只占一行的内存。

5. 写文件：w / a / x 模式

📝 w 模式 — 覆盖写入

# w = write（覆盖写入）
# 文件不存在 → 创建新文件
# 文件已存在 → 清空后重写！⚠️

with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")

# 再次用 w 写入 — 旧内容被清空！
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("新内容\n")

# output.txt 现在只有"新内容"，原来的没了

📝 a 模式 — 追加写入

# a = append（追加写入）
# 文件不存在 → 创建新文件
# 文件已存在 → 在末尾追加，不删除旧内容 ✅

with open("log.txt", "a", encoding="utf-8") as f:
    f.write("2024-01-01: 系统启动\n")

with open("log.txt", "a", encoding="utf-8") as f:
    f.write("2024-01-02: 用户登录\n")

# log.txt 现在有两行（追加，不覆盖）

📝 x 模式 — 独占创建

# x = exclusive（独占创建）
# 文件不存在 → 创建新文件
# 文件已存在 → 报错 FileExistsError！

try:
    with open("new_file.txt", "x", encoding="utf-8") as f:
        f.write("这是新文件\n")
    print("创建成功")
except FileExistsError:
    print("文件已存在，不能覆盖")

📝 writelines() — 批量写入

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

with open("fruits.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)  # 一次写入多行

# 注意：writelines 不会自动加换行符！
# 所以每个字符串末尾要自己加 \n

📋 写入模式对比

6. 读写模式大全

💡 encoding 参数：
读写中文文件时，必须指定 encoding="utf-8"。
不指定的话，Windows 默认用 GBK，中文会乱码！

7. 文件指针与 seek

文件指针就像"光标"——标记当前读/写到哪个位置。

with open("test.txt", "r", encoding="utf-8") as f:
    # tell() — 当前位置
    print(f.tell())         # 0（文件开头）

    content = f.read(5)     # 读5个字符
    print(content)          # "Hello"
    print(f.tell())         # 5（位置移动了）

    content = f.read(5)     # 继续读5个字符
    print(content)          # " Pyth"
    print(f.tell())         # 10

# seek(offset, whence) — 移动指针
with open("test.txt", "r", encoding="utf-8") as f:
    f.read()                # 读到末尾
    print(f.tell())         # 文件末尾位置

    f.seek(0)               # 回到开头
    print(f.read(5))        # "Hello"（从头开始读）

    f.seek(7)               # 跳到第7个字节
    print(f.readline())     # 从第7个字节开始读一行

# seek 的 whence 参数
# seek(offset)        → 从文件开头
# seek(offset, 0)     → 从文件开头（默认）
# seek(offset, 1)     → 从当前位置（仅二进制模式）
# seek(offset, 2)     → 从文件末尾

with open("test.txt", "rb") as f:
    f.seek(0, 2)           # 移到末尾
    size = f.tell()        # 获取文件大小
    print(f"文件大小: {size} 字节")

8. 实战练习

🎯 练习 1：通讯录管理器

import os

CONTACTS_FILE = "contacts.txt"

def load_contacts():
    """从文件加载通讯录"""
    contacts = []
    if not os.path.exists(CONTACTS_FILE):
        return contacts

    with open(CONTACTS_FILE, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                parts = line.split(",")
                if len(parts) == 3:
                    contacts.append({
                        "name": parts[0],
                        "phone": parts[1],
                        "email": parts[2],
                    })
    return contacts

def save_contacts(contacts):
    """保存通讯录到文件"""
    with open(CONTACTS_FILE, "w", encoding="utf-8") as f:
        for c in contacts:
            f.write(f"{c['name']},{c['phone']},{c['email']}\n")

def add_contact(contacts):
    """添加联系人"""
    name = input("  姓名: ").strip()
    phone = input("  电话: ").strip()
    email = input("  邮箱: ").strip()

    contacts.append({"name": name, "phone": phone, "email": email})
    save_contacts(contacts)
    print(f"  ✅ 已添加: {name}")

def search_contacts(contacts):
    """搜索联系人"""
    keyword = input("  搜索关键词: ").strip()
    results = [c for c in contacts if keyword in c["name"] or keyword in c["phone"]]

    if results:
        for c in results:
            print(f"  📇 {c['name']} | {c['phone']} | {c['email']}")
    else:
        print("  ❌ 未找到")

def list_contacts(contacts):
    """显示所有联系人"""
    if not contacts:
        print("  通讯录为空")
        return

    print(f"\n  📋 通讯录（共 {len(contacts)} 人）：")
    for i, c in enumerate(contacts, 1):
        print(f"  {i}. {c['name']:8s} | {c['phone']:12s} | {c['email']}")

# 主程序
def contact_manager():
    contacts = load_contacts()

    while True:
        print("\n📇 通讯录管理")
        print("  1. 查看  2. 添加  3. 搜索  4. 删除  q. 退出")
        choice = input("请选择: ").strip()

        if choice == "1":
            list_contacts(contacts)
        elif choice == "2":
            add_contact(contacts)
        elif choice == "3":
            search_contacts(contacts)
        elif choice == "4":
            name = input("  要删除的姓名: ").strip()
            before = len(contacts)
            contacts = [c for c in contacts if c["name"] != name]
            if len(contacts) < before:
                save_contacts(contacts)
                print(f"  ✅ 已删除: {name}")
            else:
                print(f"  ❌ 未找到: {name}")
        elif choice == "q":
            print("👋 再见！")
            break

contact_manager()

🎯 练习 2：日志分析器

from datetime import datetime
from collections import Counter

def create_sample_log():
    """创建示例日志文件"""
    logs = [
        "2024-01-15 08:30:15 [INFO] 用户登录成功 user_id=1001",
        "2024-01-15 08:31:22 [INFO] 页面访问 /home",
        "2024-01-15 08:35:45 [ERROR] 数据库连接失败 timeout=30s",
        "2024-01-15 09:00:12 [INFO] 用户登录成功 user_id=1002",
        "2024-01-15 09:15:33 [WARNING] 磁盘使用率 85%",
        "2024-01-15 09:20:01 [ERROR] API 请求超时 /api/users",
        "2024-01-15 09:25:44 [INFO] 用户登出 user_id=1001",
        "2024-01-15 10:00:00 [INFO] 定时任务开始",
        "2024-01-15 10:05:30 [ERROR] 内存不足 available=128MB",
        "2024-01-15 10:10:15 [INFO] 定时任务完成",
        "2024-01-15 10:15:00 [WARNING] 响应时间过长 avg=2.5s",
        "2024-01-15 11:00:00 [INFO] 系统备份开始",
        "2024-01-15 11:30:00 [INFO] 系统备份完成 size=2.3GB",
    ]

    with open("app.log", "w", encoding="utf-8") as f:
        for log in logs:
            f.write(log + "\n")
    print("✅ 示例日志已创建")

def analyze_log(filename):
    """分析日志文件"""
    level_count = Counter()       # 各级别计数
    error_lines = []              # 错误行
    hourly_traffic = Counter()    # 每小时流量

    with open(filename, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue

            # 解析日志级别
            if "[ERROR]" in line:
                level_count["ERROR"] += 1
                error_lines.append(line)
            elif "[WARNING]" in line:
                level_count["WARNING"] += 1
            elif "[INFO]" in line:
                level_count["INFO"] += 1

            # 解析小时
            try:
                time_str = line[:19]
                hour = int(time_str[11:13])
                hourly_traffic[hour] += 1
            except (ValueError, IndexError):
                pass

    # 输出报告
    print("\n📊 日志分析报告")
    print("=" * 50)
    print(f"日志文件: {filename}")

    print(f"\n📈 日志级别统计：")
    for level, count in level_count.most_common():
        bar = "█" * count
        print(f"  {level:8s} | {bar} ({count})")

    print(f"\n⏰ 每小时流量：")
    for hour in sorted(hourly_traffic.keys()):
        count = hourly_traffic[hour]
        bar = "█" * count
        print(f"  {hour:02d}:00 | {bar} ({count})")

    if error_lines:
        print(f"\n❌ 错误详情（共 {len(error_lines)} 条）：")
        for line in error_lines:
            print(f"  {line}")

# 运行
create_sample_log()
analyze_log("app.log")

🎯 练习 3：文件备份工具

import os
import shutil
from datetime import datetime

def backup_file(source, backup_dir="backups"):
    """
    备份文件：在文件名后加时间戳

    例如: data.txt → backups/data_20240115_143022.txt
    """
    if not os.path.exists(source):
        print(f"❌ 源文件不存在: {source}")
        return None

    # 创建备份目录
    os.makedirs(backup_dir, exist_ok=True)

    # 生成备份文件名
    name, ext = os.path.splitext(os.path.basename(source))
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_name = f"{name}_{timestamp}{ext}"
    backup_path = os.path.join(backup_dir, backup_name)

    # 复制文件
    shutil.copy2(source, backup_path)

    # 获取文件大小
    size = os.path.getsize(backup_path)
    print(f"✅ 备份成功: {backup_path} ({size} 字节)")

    return backup_path

def list_backups(backup_dir="backups"):
    """列出所有备份"""
    if not os.path.exists(backup_dir):
        print("  暂无备份")
        return

    files = os.listdir(backup_dir)
    if not files:
        print("  暂无备份")
        return

    print(f"\n  📦 备份列表（共 {len(files)} 个）：")
    for f in sorted(files):
        path = os.path.join(backup_dir, f)
        size = os.path.getsize(path)
        mtime = datetime.fromtimestamp(os.path.getmtime(path))
        print(f"    {f:40s} | {size:8d} 字节 | {mtime}")

# 使用示例
backup_file("test.txt")
backup_file("app.log")
list_backups()

9. 今日小结

🧠 记忆口诀：
open 打开 with 关，读写模式记心间。
read 读取 write 写，w 覆盖来 a 追加。
readline 读一行，readlines 列表装。
编码 utf-8 不能忘，大文件逐行最稳当。

🔮 预告： Day 17 异常处理 — try/except/else/finally。让程序不再因为一个小错误就崩溃！