Python文件操作详解:从基础到进阶

        前言

一、文件操作基础

1、打开文件

2、关闭文件

二、读取文件

1、读取全部内容

2、逐行读取

3、读取所有行到列表

三、写入文件

1、写入字符串

2、写入多行

2.1、常见使用场景

2.2、注意事项与常见错误

 2.3、高级用法

2.4、性能对比

2.5、最佳实践总结

 3、追加写入

四、文件指针操作

五、二进制文件操作

六、实际工作场景中的常见代码示例

1、日志文件处理(按条件过滤)

2、CSV数据清洗(处理缺失值)

3、大文件分块读取(内存优化)

4、配置文件解析(JSON/YAML)

5、文件监控(实时处理新内容)

 6、多文件合并(归并处理)

前言

最近在学习Python,将所学内容整理成文章,以便以后翻阅和复习,同时分享给大家。有问题欢迎随时指正。

一、文件操作基础

1、打开文件

open(file_path, mode, encoding) 

  • file_path:文件路径,可以是相对路径或绝对路径。
  • mode:打开模式,常用模式包括:
  • 模式 描述
    r 只读(默认)
    w 写入(覆盖原文件)
    a 追加写入
    x 创建新文件并写入
    b 二进制模式(如 rbwb
    + 读写模式(如 r+w+
    # 示例:打开文件(推荐使用 with 自动管理资源)
    with open('example.txt', 'r', encoding='utf-8') as f:
        content = f.read()

    2、关闭文件

    close()

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

    对比with语句(推荐方式) 

    # 使用 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

    二、读取文件

    1、读取全部内容

    read()

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

    2、逐行读取

    readline()

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

    3、读取所有行到列表

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

    三、写入文件

    1、写入字符串

     write(text)

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

    2、写入多行

    writelines(lines) 

    lines = ['Line 1\n', 'Line 2\n']
    with open('output.txt', 'w') as f:
        f.writelines(lines)  # 写入列表中的每行内容
    2.1、常见使用场景

    场景1:从其他数据源生成行 

    # 将数值列表转为文本行
    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

    场景2:追加模式写入

    # 在已有文件后追加新行
    new_lines = ["追加行1\n", "追加行2\n"]
    
    with open('existing_file.txt', 'a', encoding='utf-8') as f:
        f.writelines(new_lines)

    场景3:动态生成内容

    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
    2.2、注意事项与常见错误

    错误1:忘记换行符

    # 错误写法:所有内容挤在一行
    lines = ["内容A", "内容B", "内容C"]
    with open('wrong.txt', 'w') as f:
        f.writelines(lines)  # 输出:内容A内容B内容C
    
    # 正确写法:
    lines = ["内容A\n", "内容B\n", "内容C\n"]

    错误2:非字符串类型

    # 错误案例:包含非字符串元素
    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])
     2.3、高级用法

    方法1:配合生成器处理大文件

    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))

     方法2:二进制模式写入

    # 写入二进制换行符(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)
    2.4、性能对比
    方法 10万行耗时 内存占用 适用场景
    write() 循环 0.45s 需要逐行处理
    writelines() 0.12s 批量写入
    生成器 + writelines() 0.15s 极低 超大文件
    2.5、最佳实践总结
    1. 始终添加换行符writelines 不会自动添加换行

    2. 类型统一:确保列表元素都是字符串类型

    3. 大文件优化:使用生成器替代列表存储所有行

    4. 模式选择

    5. 'w':覆盖写入

    6. 'a':追加写入

    7. 搭配使用:常与列表推导式结合快速生成内容

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

     3、追加写入

    使用模式 a

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

    四、文件指针操作

    1. 获取当前位置tell()

    2. 移动指针seek(offset, whence)

    3. whence:0(文件开头),1(当前位置),2(文件末尾)

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

    五、二进制文件操作

     使用 b 模式处理非文本文件(如图片、视频):

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

    六、实际工作场景中的常见代码示例

    1、日志文件处理(按条件过滤)

    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')

    2、CSV数据清洗(处理缺失值)

    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')

    3、大文件分块读取(内存优化)

    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}")

    4、配置文件解析(JSON/YAML)

    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'])

    5、文件监控(实时处理新内容)

    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}")

     6、多文件合并(归并处理)

    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')

    作者:三生暮雨渡瀟瀟

    物联沃分享整理
    物联沃-IOTWORD物联网 » Python文件操作详解:从基础到进阶

    发表回复