文件操作

# 打开文件（只读模式）
file = open('example.txt', 'r')

# 读取文件内容
content = file.read()
print(content)

# 关闭文件
file.close()

# --- 'r' 只读模式：最常用，文件必须存在 ---
with open('config.txt', 'r', encoding='utf-8') as f:
    content = f.read()  # 读取全部内容

# --- 'w' 写入模式：会清空已有内容！谨慎使用 ---
with open('output.txt', 'w', encoding='utf-8') as f:
    f.write('这会覆盖文件原有内容')  # 原有内容全部丢失

# --- 'a' 追加模式：在文件末尾添加，不影响已有内容 ---
with open('app.log', 'a', encoding='utf-8') as f:
    f.write('2026-06-04 新的日志条目\n')  # 追加到末尾

# --- 'r+' 读写模式：可读可写，文件必须存在，写入从指针位置开始 ---
with open('data.txt', 'r+', encoding='utf-8') as f:
    old = f.read()       # 先读取（指针移到末尾）
    f.write('追加内容')  # 在当前指针位置（末尾）写入

# --- 'x' 独占创建模式：文件已存在则报错，防止意外覆盖 ---
try:
    with open('new_file.txt', 'x', encoding='utf-8') as f:
        f.write('安全创建的新文件')
except FileExistsError:
    print('文件已存在，不会覆盖！')

# --- 'b' 二进制模式：处理非文本文件 ---
with open('photo.jpg', 'rb') as f:   # 二进制读取
    image_data = f.read()
with open('copy.jpg', 'wb') as f:    # 二进制写入
    f.write(image_data)

# --- 't' 文本模式（默认）：与 'b' 相对 ---
# open('file.txt', 'r') 等价于 open('file.txt', 'rt')

# 指定编码格式读取文件
with open('example_utf8.txt', 'r', encoding='utf-8') as file:
    content = file.read()

# 指定编码格式写入文件
with open('example_gbk.txt', 'w', encoding='gbk') as file:
    file.write('这是一个测试。')

# errors 参数的常用取值：
# 'strict'  — 默认，遇到错误抛出 UnicodeDecodeError
# 'ignore'  — 忽略无法解码的字节（可能丢失数据）
# 'replace' — 用 � 替换无法解码的字节
# 'backslashreplace' — 用 \xNN 转义序列替换

# 场景：读取一个混合编码的日志文件，部分字节无法解码
with open('mixed_encoding.log', 'r', encoding='utf-8', errors='replace') as f:
    content = f.read()
    # 无法解码的字节会被替换为 �，程序不会崩溃

# 场景：需要知道哪些字节有问题，用转义序列标记
with open('mixed_encoding.log', 'r', encoding='utf-8', errors='backslashreplace') as f:
    content = f.read()
    # 无法解码的字节会显示为 \xff 这样的转义序列

# 使用 with 语句打开文件
with open('example.txt', 'r') as file:
    content = file.read()
    print(content)

# 文件已自动关闭，无需显式调用 file.close()

# 同时读写两个文件
with open('input.txt', 'r', encoding='utf-8') as fin, \
        open('output.txt', 'w', encoding='utf-8') as fout:
    for line in fin:
        fout.write(line.upper())

class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()

# 使用自定义上下文管理器
with FileManager('example_custom.txt', 'w') as file:
    file.write('Using a custom context manager.')

from pathlib import Path
import os

# 从字符串创建 Path 对象
p = Path('/Users/zhangzhengyang/Desktop/终结者 python 笔记/py-vitepress/docs/基础知识/13-文件操作.md')

# 获取当前工作目录
cwd = Path.cwd()
print(f"当前目录: {cwd}")

# 获取用户主目录
home = Path.home()
print(f"主目录: {home}")

from pathlib import Path

# 拼接路径
data_dir = Path.home() / 'data'
report_path = data_dir / 'reports' / '2025' / 'report.csv'
print(f"报告路径: {report_path}")

# 访问路径的各个部分
print(f"  - 父目录: {report_path.parent}")
print(f"  - 文件名: {report_path.name}")
print(f"  - 文件名（无后缀）: {report_path.stem}")
print(f"  - 后缀: {report_path.suffix}")
print(f"  - 各部分组成的元组: {report_path.parts}")

from pathlib import Path

p = Path('students.csv')

print(f"路径 '{p}' 是否存在? {p.exists()}")
print(f"路径 '{p}' 是文件吗? {p.is_file()}")
print(f"路径 '{p}' 是目录吗? {p.is_dir()}")

from pathlib import Path

# 创建单级目录
results_dir = Path('results')
results_dir.mkdir(exist_ok=True)  # exist_ok=True: 如果目录已存在，不抛出错误

# 创建多级目录
archive_dir = Path('归档/2025/monthly')
archive_dir.mkdir(parents=True, exist_ok=True) # parents=True: 自动创建所有父目录

from pathlib import Path

p = Path('.') # 当前目录
for item in p.iterdir():
    if item.is_dir():
        print(f"目录: {item.name}")
    else:
        print(f"文件: {item.name}")

from pathlib import Path

docs_dir = Path('/Users/zhangzhengyang/Desktop/终结者 python 笔记/py-vitepress/docs')

# 查找所有 Markdown 文件
md_files = list(docs_dir.glob('*.md'))
print(f"找到的 Markdown 文件: {md_files}")

# 递归查找所有 Python 文件
py_files_recursive = list(docs_dir.rglob('*.py'))
print(f"递归找到的 Python 文件: {py_files_recursive}")

from pathlib import Path

p = Path('greeting.txt')

# 写入文本，自动处理文件打开和关闭
p.write_text('Hello, pathlib!', encoding='utf-8')

# 读取文本
content = p.read_text(encoding='utf-8')
print(content)

# 写入二进制数据
p_bin = Path('data.bin')
p_bin.write_bytes(b'\x00\x01\x02')

# 读取二进制数据
binary_content = p_bin.read_bytes()
print(binary_content)

# --- read()：一次性读取全部内容 ---
# 适合小文件（如配置文件），大文件会占用大量内存
with open('small_config.txt', 'r', encoding='utf-8') as f:
    all_content = f.read()  # 返回整个文件的字符串
    print(f"文件共 {len(all_content)} 个字符")

# --- readline()：每次读取一行 ---
# 适合需要精确控制读取进度的场景
with open('data.txt', 'r', encoding='utf-8') as f:
    first_line = f.readline()   # 读取第一行
    second_line = f.readline()  # 读取第二行
    # readline() 返回空字符串 '' 表示文件结束（不是 None）

# --- readlines()：读取所有行到列表 ---
# 适合需要随机访问某一行，或多次遍历的场景
# 注意：大文件会占用大量内存
with open('data.txt', 'r', encoding='utf-8') as f:
    lines = f.readlines()  # 返回列表，每个元素是一行（含 \n）
    print(f"共 {len(lines)} 行")
    # 可以通过索引随机访问：lines[5] 是第 6 行

# --- 迭代（推荐）：逐行读取，内存最省 ---
# 文件对象本身是可迭代的，每次 yield 一行
# 这是处理大文件的最佳方式
with open('large_log.txt', 'r', encoding='utf-8') as f:
    for line_number, line in enumerate(f, 1):  # 行号从 1 开始
        if 'ERROR' in line:
            print(f"第 {line_number} 行发现错误: {line.strip()}")

from pathlib import Path

path = Path('my_large_log.txt')

# 假设文件内容是：
# INFO: Task started
# WARNING: Deprecated feature used
# ERROR: Connection failed

with path.open('r', encoding='utf-8') as f:
    for line in f:
        # .strip() 用于移除行尾的换行符 \n
        if "ERROR" in line:
            print(f"发现错误: {line.strip()}")

from pathlib import Path

path = Path('config.ini')

# 假设文件内容是：
# [database]
# host = localhost

try:
    content = path.read_text(encoding='utf-8')
    print("配置文件内容：\n", content)
except FileNotFoundError:
    print(f"错误: 文件 '{path}' 未找到。")

from pathlib import Path

log_path = Path('application.log')

# 写入模式 ('w')：清空并写入新日志
log_path.write_text("--- 日志开始 ---\n", encoding='utf-8')

# 追加模式 ('a')：添加更多日志条目
with log_path.open('a', encoding='utf-8') as f:
    f.write("2025-11-27 10:00 - 应用启动\n")
    f.write("2025-11-27 10:01 - 连接数据库...\n")

print(log_path.read_text(encoding='utf-8'))

# --- write()：写入字符串 ---
# 返回写入的字符数（文本模式）或字节数（二进制模式）
with open('output.txt', 'w', encoding='utf-8') as f:
    chars_written = f.write('Hello, World!\n')  # 返回 14
    print(f"写入了 {chars_written} 个字符")

# --- writelines()：写入字符串列表 ---
# 注意：writelines 不会自动添加换行符！需要手动在每行末尾加 \n
lines = ['第一行\n', '第二行\n', '第三行\n']
with open('multi_lines.txt', 'w', encoding='utf-8') as f:
    f.writelines(lines)  # 一次性写入多行

# 对比：逐行 write vs writelines
# 逐行 write（每次都触发缓冲区写入）
with open('slow.txt', 'w', encoding='utf-8') as f:
    for line in lines:
        f.write(line)

# writelines（一次性提交，效率更高）
with open('fast.txt', 'w', encoding='utf-8') as f:
    f.writelines(lines)

from pathlib import Path

source_image = Path('logo.png')
destination_image = Path('logo_copy.png')

# 假设 logo.png 存在于当前目录
if not source_image.exists():
    print(f"源文件 '{source_image}' 不存在！")
else:
    try:
        # 以二进制读取模式 ('rb') 打开源文件
        binary_data = source_image.read_bytes()

        # 以二进制写入模式 ('wb') 写入目标文件
        destination_image.write_bytes(binary_data)

        print(f"图片已成功复制到 '{destination_image}'")

    except Exception as e:
        print(f"复制文件时发生错误: {e}")

# 对于大文件（如视频、数据库备份），不要一次性 read() 全部内容
# 使用分块读写，控制内存使用
def copy_large_file(src: str, dst: str, chunk_size: int = 64 * 1024) -> None:
    """分块复制大文件，每次只读 chunk_size 字节到内存"""
    with open(src, 'rb') as f_in, open(dst, 'wb') as f_out:
        while True:
            chunk = f_in.read(chunk_size)  # 每次读取 64KB
            if not chunk:                  # 读到文件末尾，chunk 为空 bytes
                break
            f_out.write(chunk)             # 写入当前块

# 使用示例：复制一个 2GB 的视频文件，内存占用始终只有 64KB
copy_large_file('presentation.mp4', 'presentation_backup.mp4')

import json
from pathlib import Path

config_path = Path('app_config.json')

# 创建一个配置字典
config_data = {
    'database': {
        'host': 'localhost',
        'port': 5432,
        'user': 'admin'
    },
    'api_keys': [
        {'service': 'google', 'key': 'AIzaSy...'},
        {'service': 'stripe', 'key': 'sk_test_...'}
    ],
    'debug_mode': True
}

# --- 写入 JSON 文件 ---
try:
    with config_path.open('w', encoding='utf-8') as f:
        # indent=4 使文件格式化，更易读
        # ensure_ascii=False 确保中文字符能被正确写入，而不是被转义为 \uXXXX
        json.dump(config_data, f, indent=4, ensure_ascii=False)
    print(f"配置已成功写入到 '{config_path}'")

except Exception as e:
    print(f"写入 JSON 文件时出错: {e}")

# --- 读取 JSON 文件 ---
try:
    with config_path.open('r', encoding='utf-8') as f:
        loaded_config = json.load(f)
    print("\n成功加载配置:")
    print(f"  数据库主机: {loaded_config['database']['host']}")
    print(f"  第一个 API 服务: {loaded_config['api_keys'][0]['service']}")

except FileNotFoundError:
    print(f"错误: 配置文件 '{config_path}' 不存在。")
except json.JSONDecodeError:
    print(f"错误: 配置文件 '{config_path}' 格式不正确。")
except Exception as e:
    print(f"读取 JSON 文件时出错: {e}")

import csv
from pathlib import Path

students_path = Path('students.csv')

# 待写入的数据：一个字典列表
student_data = [
    {'name': '张三', 'math_score': 92, 'english_score': 88},
    {'name': '李四', 'math_score': 78, 'english_score': 95},
    {'name': '王五', 'math_score': 85, 'english_score': 89}
]

# --- 写入 CSV 文件 ---
try:
    with students_path.open('w', newline='', encoding='utf-8') as f:
        # 定义表头
        fieldnames = ['name', 'math_score', 'english_score']

        # 创建一个 DictWriter 对象
        writer = csv.DictWriter(f, fieldnames=fieldnames)

        # 写入表头
        writer.writeheader()

        # 写入所有数据行
        writer.writerows(student_data)

    print(f"学生数据已成功写入到 '{students_path}'")

except Exception as e:
    print(f"写入 CSV 文件时出错: {e}")

# --- 读取 CSV 文件 ---
try:
    with students_path.open('r', encoding='utf-8') as f:
        # 创建一个 DictReader 对象
        reader = csv.DictReader(f)

        print("\n从 CSV 文件中读取的学生数据:")
        for row in reader:
            # row 是一个字典，可以直接通过列名访问
            print(f"  姓名: {row['name']}, 数学: {row['math_score']}, 英语: {row['english_score']}")

except FileNotFoundError:
    print(f"错误: CSV 文件 '{students_path}' 不存在。")
except Exception as e:
    print(f"读取 CSV 文件时出错: {e}")

with open('example.txt', 'r') as file:
    content = file.read(10)
    position = file.tell()
    print(f'已读取 10 个字符，当前指针位置: {position}')

with open('example.txt', 'r') as file:
    file.seek(5)  # 移动到第 6 个字节
    content = file.read(10)
    print(content)

with open('image.jpg', 'rb') as file:
    binary_data = file.read()
    # 可以对 binary_data 进行处理，如保存到另一个文件

data_to_write = b'\x00\x01\x02\x03'  # 字节数据
with open('binary_output.bin', 'wb') as file:
    file.write(data_to_write)

with open('example_flush.txt', 'w') as file:
    file.write('Data to flush')
    file.flush()  # 确保数据立即写入磁盘

with open('example.txt', 'r') as file:
    if file.isatty():
        print("文件连接到一个终端设备。")
    else:
        print("文件未连接到一个终端设备。")  # 通常为这种情况

with open('example_truncate.txt', 'w+') as file:
    file.write('This is a longer text that will be truncated.')
    file.seek(10)  # 移动指针到第 11 个字节
    file.truncate()  # 截断文件，保留前 10 个字节

try:
    with open('nonexistent_file.txt', 'r') as file:
        content = file.read()
except FileNotFoundError:
    print("错误：文件未找到。")

try:
    with open('example.txt', 'r') as file:
        content = file.read(100)
        # 假设对内容进行一些处理，可能引发 ValueError
        if 'error' in content:
            raise ValueError("内容中包含错误关键字。")
except FileNotFoundError:
    print("错误：文件未找到。")
except PermissionError:
    print("错误：没有权限访问文件。")
except ValueError as ve:
    print(f"值错误：{ve}")
except Exception as e:
    print(f"发生未知错误：{e}")

with open('large_file.txt', 'r') as file:
    for line in file:
        # 处理每一行
        process(line.strip())

chunk_size = 1024  # 每次读取 1KB
with open('large_file.bin', 'rb') as file:
    while True:
        chunk = file.read(chunk_size)
        if not chunk:
            break
        # 处理每个块
        process_chunk(chunk)

import mmap
from pathlib import Path

# 场景：在一个 10GB 的日志文件中搜索特定关键字
# 使用 mmap，内存占用只有几 KB，而不是 10GB
log_path = Path('huge_server.log')

with log_path.open('r', encoding='utf-8') as f:
    # 将文件映射到内存，access=mmap.ACCESS_READ 表示只读
    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        # mmap 对象可以像 bytes 一样操作，但不会全部加载到内存
        # 搜索关键字（比逐行读取快得多）
        keyword = b'CRITICAL ERROR'  # mmap 操作的是 bytes
        position = mm.find(keyword)
        if position != -1:
            # 定位到关键字所在位置，读取周围上下文
            mm.seek(max(0, position - 100))  # 向前 100 字节
            context = mm.read(200 + len(keyword))  # 读取 200+ 字节的上下文
            print(f"在位置 {position} 找到关键字")
            print(f"上下文: {context.decode('utf-8', errors='replace')}")
        else:
            print("未找到关键字")

# mmap 的优势：
# 1. 内存占用极低——OS 按需加载页面，不是全部读入
# 2. 支持随机访问——seek 到任意位置，不需要从头读
# 3. 多进程共享——多个进程可以映射同一文件，共享数据

import tempfile
import shutil
from pathlib import Path

target = Path('report.txt')
with tempfile.NamedTemporaryFile('w', delete=False, encoding='utf-8') as tmp:
    tmp.write('new content')
    tmp_path = Path(tmp.name)

shutil.move(str(tmp_path), target)  # 原子替换（同一文件系统内）

import tempfile
from pathlib import Path

# --- NamedTemporaryFile：创建有名字的临时文件 ---
# delete=False：文件关闭后不会自动删除，需要手动处理
# 适合"先写临时文件，再原子替换目标文件"的安全写入模式
with tempfile.NamedTemporaryFile(mode='w', delete=False, encoding='utf-8', suffix='.json') as tmp:
    tmp.write('{"status": "ok"}')
    tmp_path = Path(tmp.name)
    print(f"临时文件路径: {tmp_path}")
# 文件仍然存在，可以用于 shutil.move() 原子替换

# --- TemporaryFile：创建匿名临时文件 ---
# 文件关闭后自动删除，适合中间处理步骤
with tempfile.TemporaryFile(mode='w+', encoding='utf-8') as tmp:
    tmp.write('中间数据\n')
    tmp.seek(0)              # 回到文件开头
    content = tmp.read()     # 读取刚写入的内容
    print(f"中间数据: {content.strip()}")
# 文件已自动删除，无需手动清理

# --- TemporaryDirectory：创建临时目录 ---
# 适合需要多个临时文件的场景（如解压、批量处理）
with tempfile.TemporaryDirectory() as tmp_dir:
    tmp_dir_path = Path(tmp_dir)
    # 在临时目录中创建多个文件
    (tmp_dir_path / 'part1.txt').write_text('数据1', encoding='utf-8')
    (tmp_dir_path / 'part2.txt').write_text('数据2', encoding='utf-8')
    # 列出临时目录中的文件
    for f in tmp_dir_path.iterdir():
        print(f"临时文件: {f.name}")
# 退出 with 块后，整个临时目录及其内容自动删除

from pathlib import Path

def process_line(line: str) -> str:
    """示例处理函数：将行转换为大写"""
    return line.upper()

def main() -> None:
    input_path = Path('input.txt')
    output_path = Path('output.txt')

    # 检查输入文件是否存在
    if not input_path.exists():
        print(f"错误：输入文件 '{input_path}' 不存在。")
        return

    try:
        with input_path.open('r', encoding='utf-8') as fin, \
                output_path.open('w', encoding='utf-8') as fout:
            for line_number, line in enumerate(fin, 1):
                processed_line = process_line(line.rstrip('\n'))
                fout.write(f"{line_number}: {processed_line}\n")
        print(f"处理完成，结果已写入 '{output_path}'。")
    except OSError as e:
        print(f"文件操作错误：{e}")
    except Exception as e:
        print(f"发生未知错误：{e}")

if __name__ == '__main__':
    main()

# ❌ 错误：忘记关闭文件
f = open('data.txt', 'r')
content = f.read()
# 忘记 f.close()！
# 后果：文件描述符泄漏，长期运行的服务可能耗尽系统资源
# 在 Windows 上，未关闭的文件会被锁定，其他程序无法访问

# ✅ 正确：使用 with 语句，自动关闭
with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()
# 离开 with 块后，文件自动关闭，即使发生异常也是如此

# ❌ 错误：不指定编码，依赖系统默认值
with open('data.txt', 'r') as f:  # Windows 默认 gbk，Linux 默认 utf-8
    content = f.read()
# 同一文件在不同系统上可能产生乱码或 UnicodeDecodeError

# ✅ 正确：始终显式指定 encoding
with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()

# ✅ 遇到编码不明的文件，使用 errors 参数容错
with open('unknown_encoding.txt', 'r', encoding='utf-8', errors='replace') as f:
    content = f.read()  # 无法解码的字节替换为 �，不会崩溃

# ❌ 错误：对大文件使用 read() 或 readlines()
with open('10GB_database_dump.txt', 'r') as f:
    all_lines = f.readlines()  # 尝试将 10GB 加载到内存！
# 后果：MemoryError，程序崩溃，甚至系统卡死

# ✅ 正确：逐行迭代，内存占用恒定
with open('10GB_database_dump.txt', 'r', encoding='utf-8') as f:
    for line in f:
        if 'target_keyword' in line:
            process(line.strip())
            break  # 找到后可以提前退出，不用读完整个文件

# ✅ 二进制大文件：分块读取
CHUNK_SIZE = 64 * 1024  # 64KB
with open('large_video.mp4', 'rb') as f:
    while chunk := f.read(CHUNK_SIZE):  # 海象运算符，Python 3.8+
        process_chunk(chunk)

# ❌ 错误：硬编码路径分隔符
config_path = 'data\\config\\settings.json'  # Windows 路径
# 在 Linux/macOS 上会失败，因为它们使用 / 作为分隔符

# ❌ 错误：手动拼接路径
data_dir = base_dir + '/' + subdir + '/' + filename  # 不安全，可能产生双斜杠

# ✅ 正确：使用 pathlib，自动处理跨平台分隔符
from pathlib import Path
config_path = Path('data') / 'config' / 'settings.json'
# Windows 上自动变为 data\config\settings.json
# Linux/macOS 上自动变为 data/config/settings.json

# ✅ 正确：使用 os.path.join（传统方式）
import os
config_path = os.path.join('data', 'config', 'settings.json')

# Windows 使用 \r\n（CRLF）作为换行符
# Linux/macOS 使用 \n（LF）作为换行符
# 这可能导致意想不到的问题：

# ❌ 问题：在 Windows 上创建的文件，在 Linux 上读取时行尾多出 \r
with open('windows_file.txt', 'r', encoding='utf-8') as f:
    for line in f:
        # line 末尾可能是 '...\r\n' 而不是 '...\n'
        # 如果用 line == 'expected_text\n' 比较，会失败
        print(repr(line))  # 可能显示 '...\r\n'

# ✅ 解决方案 1：使用 .strip() 移除所有空白字符（包括 \r）
with open('windows_file.txt', 'r', encoding='utf-8') as f:
    for line in f:
        clean_line = line.strip()  # 移除首尾空白，包括 \r 和 \n

# ✅ 解决方案 2：文本模式自动转换（默认行为）
# Python 文本模式在读取时自动将平台换行符转换为 \n
# 在写入时自动将 \n 转换为平台换行符
# 所以大多数情况下，你不需要手动处理 \r\n

# ✅ 解决方案 3：如果需要保留原始换行符，使用 newline=''
with open('raw_file.txt', 'r', encoding='utf-8', newline='') as f:
    for line in f:
        # line 保留原始换行符，不做任何转换
        print(repr(line))