Python学习教程之文件操作详解

# 示例：打开文件（推荐使用 with 自动管理资源）
with open('example.txt', 'r', encoding='utf-8') as f:
    content = f.read()

# 手动打开并关闭文件（不推荐，容易忘记关闭）
f = open('demo.txt', 'w', encoding='utf-8')
f.write('hello world')
f.close()  # 必须显式关闭

# 使用 with 自动关闭（最佳实践）
with open('demo.txt', 'w') as f:
    f.write('auto-closed file')

# 等效于：
f = open('demo.txt', 'w')
try:
    f.write('auto-closed file')
finally:
    f.close()

# 案例：文件未关闭导致写入失败
def write_data():
    f = open('data.txt', 'w')
    f.write('important data')
    # 忘记 f.close() → 数据可能仍在缓冲区未写入磁盘

# 正确写法：
def safe_write():
    with open('data.txt', 'w') as f:
        f.write('saved data')  # with 块结束自动关闭并写入磁盘

# 立即读取刚写入的内容
with open('data.txt', 'w') as f:
    f.write('test')
    
with open('data.txt', 'r') as f:  # 必须重新打开才能读取新内容
    print(f.read())  # 输出 test

with open('example.txt', 'r') as f:
    data = f.read()  # 返回整个文件内容的字符串

with open('example.txt', 'r') as f:
    line = f.readline()  # 读取一行
    while line:
        print(line.strip())  # 去除末尾换行符
        line = f.readline()

with open('example.txt', 'r') as f:
    lines = f.readlines()  # 返回包含所有行的列表
    for line in lines:
        print(line.strip())

with open('output.txt', 'w') as f:
    f.write('hello, world!\n')  # 写入内容（覆盖模式）

lines = ['line 1\n', 'line 2\n']
with open('output.txt', 'w') as f:
    f.writelines(lines)  # 写入列表中的每行内容

# 将数值列表转为文本行
numbers = [1, 2, 3, 4, 5]
lines = [f"数值: {n}\n" for n in numbers]

with open('data.log', 'w') as f:
    f.writelines(lines)

# 输出：
# 数值: 1
# 数值: 2
# 数值: 3
# 数值: 4
# 数值: 5

# 在已有文件后追加新行
new_lines = ["追加行1\n", "追加行2\n"]

with open('existing_file.txt', 'a', encoding='utf-8') as f:
    f.writelines(new_lines)

import datetime

# 生成带时间戳的日志行
timestamps = [
    f"[{datetime.datetime.now().strftime('%y-%m-%d %h:%m:%s')}] 事件 {i}\n"
    for i in range(3)
]

with open('events.log', 'w') as f:
    f.writelines(timestamps)

# 输出示例：
# [2023-08-20 09:15:23] 事件 0
# [2023-08-20 09:15:23] 事件 1
# [2023-08-20 09:15:23] 事件 2

# 错误写法：所有内容挤在一行
lines = ["内容a", "内容b", "内容c"]
with open('wrong.txt', 'w') as f:
    f.writelines(lines)  # 输出：内容a内容b内容c

# 正确写法：
lines = ["内容a\n", "内容b\n", "内容c\n"]

# 错误案例：包含非字符串元素
mixed_data = ["文本\n", 123, true]
with open('error.txt', 'w') as f:
    # f.writelines(mixed_data)  # 将抛出 typeerror
    
    # 解决方案：转换为字符串
    f.writelines([str(item)+'\n' for item in mixed_data])

def generate_large_data(num_lines):
    """生成大文件数据"""
    for i in range(num_lines):
        yield f"这是第 {i+1} 行数据\n"  # 使用生成器避免内存爆炸

# 写入100万行（内存友好）
with open('bigfile.txt', 'w') as f:
    f.writelines(generate_large_data(1_000_000))

# 写入二进制换行符（windows换行符为\r\n）
lines = [
    b"binary line 1\r\n",
    b"binary line 2\r\n"
]

with open('binary_file.bin', 'wb') as f:
    f.writelines(lines)

# 典型安全写法示例
data = [f"处理结果: {x}\n" for x in some_dataset]
with open('result.txt', 'w') as f:
    f.writelines(data)

with open('output.txt', 'a') as f:
    f.write('appended line.\n')

with open('example.txt', 'r') as f:
    f.seek(5)  # 移动指针到第5字节
    print(f.read(3))  # 读取接下来3个字符
    print(f.tell())   # 输出当前指针位置

# 复制图片
with open('input.jpg', 'rb') as src, open('copy.jpg', 'wb') as dst:
    data = src.read()
    dst.write(data)

import re
from pathlib import path

def filter_logs(input_file, output_file, keyword, start_time=none):
    """
    从日志文件中提取包含关键字的行，并可选时间范围
    :param input_file: 输入日志文件路径
    :param output_file: 输出结果文件路径
    :param keyword: 需要过滤的关键字
    :param start_time: 起始时间（格式：2023-08-01 10:00:00）
    """
    input_path = path(input_file)
    if not input_path.exists():
        raise filenotfounderror(f"日志文件 {input_file} 不存在")

    pattern = re.compile(r'\[(.*?)\]')  # 假设日志时间格式为 [2023-08-01 10:00:00]
    
    with open(input_file, 'r', encoding='utf-8') as f_in, \
         open(output_file, 'w', encoding='utf-8') as f_out:
        
        for line in f_in:
            # 时间戳提取
            time_match = pattern.search(line)
            if not time_match:
                continue
            
            log_time = time_match.group(1)
            # 时间条件判断
            if start_time and log_time < start_time:
                continue
            
            # 关键字匹配
            if keyword in line:
                f_out.write(line)

# 使用示例
filter_logs('app.log', 'error_logs.txt', 'error', '2023-08-01 12:00:00')

import csv

def clean_csv(input_file, output_file, default_value='n/a'):
    """
    清洗csv文件：处理空值并保存新文件
    """
    with open(input_file, 'r', newline='', encoding='utf-8') as f_in, \
         open(output_file, 'w', newline='', encoding='utf-8') as f_out:
        
        reader = csv.dictreader(f_in)
        writer = csv.dictwriter(f_out, fieldnames=reader.fieldnames)
        writer.writeheader()
        
        for row in reader:
            cleaned_row = {
                key: value if value.strip() != '' else default_value
                for key, value in row.items()
            }
            writer.writerow(cleaned_row)

# 使用示例
clean_csv('dirty_data.csv', 'cleaned_data.csv', default_value='0')

def process_large_file(file_path, chunk_size=1024*1024):  # 默认1mb块
    """
    分块读取大文件，避免内存溢出
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        while true:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            # 在此处处理数据块（例如：计数、分析等）
            yield chunk

# 使用示例（统计文件行数）
line_count = 0
for chunk in process_large_file('huge_file.log'):
    line_count += chunk.count('\n')

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

import json
import yaml  # 需要安装pyyaml: pip install pyyaml

def load_config(config_path):
    """
    自动识别json/yaml配置文件并加载
    """
    config_path = path(config_path)
    if not config_path.exists():
        raise filenotfounderror("配置文件不存在")
    
    with open(config_path, 'r', encoding='utf-8') as f:
        if config_path.suffix == '.json':
            return json.load(f)
        elif config_path.suffix in ('.yaml', '.yml'):
            return yaml.safe_load(f)
        else:
            raise valueerror("不支持的配置文件格式")

# 使用示例
config = load_config('app_config.yaml')
print(config['database']['host'])

import time

def tail_file(file_path, interval=1):
    """
    模拟linux tail -f 功能，实时监控文件新增内容
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        # 移动到文件末尾
        f.seek(0, 2)
        
        while true:
            line = f.readline()
            if not line:
                time.sleep(interval)
                continue
            yield line

# 使用示例（监控日志并报警）
for new_line in tail_file('app.log'):
    if 'critical' in new_line:
        send_alert(f"发现关键错误: {new_line}")

from pathlib import path

def merge_csv_files(input_dir, output_file):
    """
    合并目录下所有csv文件（假设结构相同）
    """
    input_dir = path(input_dir)
    csv_files = list(input_dir.glob('*.csv'))
    
    with open(output_file, 'w', newline='', encoding='utf-8') as f_out:
        header_written = false
        for csv_file in csv_files:
            with open(csv_file, 'r', newline='', encoding='utf-8') as f_in:
                reader = csv.reader(f_in)
                header = next(reader)
                
                if not header_written:
                    csv.writer(f_out).writerow(header)
                    header_written = true
                
                for row in reader:
                    csv.writer(f_out).writerow(row)

# 使用示例
merge_csv_files('daily_reports', 'combined_report.csv')