Python文件读写的几种方法与示例详解

# 基本语法
file_object = open(filename, mode, encoding)

# 示例1：读取整个文件内容
def read_entire_file():
    try:
        with open('example.txt', 'r', encoding='utf-8') as file:
            content = file.read()
            print("文件全部内容：")
            print(content)
    except FileNotFoundError:
        print("文件不存在")

# 示例2：逐行读取文件
def read_line_by_line():
    with open('example.txt', 'r', encoding='utf-8') as file:
        line_number = 1
        for line in file:
            print(f"第{line_number}行: {line.strip()}")
            line_number += 1

# 示例3：读取所有行到列表
def read_all_lines():
    with open('example.txt', 'r', encoding='utf-8') as file:
        lines = file.readlines()
        print(f"文件共有{len(lines)}行")
        for i, line in enumerate(lines, 1):
            print(f"第{i}行: {line.strip()}")

# 示例4：分块读取大文件
def read_large_file_in_chunks(file_path, chunk_size=1024):
    """分块读取大文件，避免内存溢出"""
    with open(file_path, 'r', encoding='utf-8') as file:
        while True:
            chunk = file.read(chunk_size)
            if not chunk:
                break
            # 处理每个数据块
            process_chunk(chunk)

def process_chunk(chunk):
    """处理数据块的示例函数"""
    print(f"处理数据块，长度：{len(chunk)}")
    # 这里可以添加具体的数据处理逻辑

# 示例5：写入单行内容
def write_single_line():
    with open('output.txt', 'w', encoding='utf-8') as file:
        file.write("这是第一行内容
")
        file.write("这是第二行内容
")
    print("文件写入完成")

# 示例6：写入多行内容
def write_multiple_lines():
    lines = [
        "第一行：Python文件操作",
        "第二行：使用write方法",
        "第三行：文件写入完成"
    ]
    
    with open('multi_lines.txt', 'w', encoding='utf-8') as file:
        for line in lines:
            file.write(line + '
')
    print("多行内容写入完成")

# 示例7：追加内容到文件
def append_to_file():
    with open('existing_file.txt', 'a', encoding='utf-8') as file:
        file.write("这是追加的新内容
")
        file.write(f"追加时间：{datetime.now()}
")
    print("内容追加完成")

# 示例8：带序号的文件写入
def write_with_line_numbers():
    content = [
        "Python是一门强大的编程语言",
        "文件操作是编程中的基础技能",
        "掌握文件读写很重要"
    ]
    
    with open('numbered_content.txt', 'w', encoding='utf-8') as file:
        for i, line in enumerate(content, 1):
            file.write(f"{i}. {line}
")
    print("带序号的内容写入完成")

# 示例9：在指定位置插入内容
def insert_content_at_position():
    """在文件特定位置插入内容"""
    with open('target_file.txt', 'r+', encoding='utf-8') as file:
        content = file.read()
        file.seek(0)  # 回到文件开头
        file.write("插入的新内容
")
        file.write(content)
    print("内容插入完成")

# 示例10：使用with上下文管理器
def file_operations_with_context():
    """使用with语句自动管理文件资源"""
    try:
        with open('data.txt', 'r', encoding='utf-8') as file:
            data = file.read()
            print("文件读取成功")
            
        # 文件会自动关闭，无需手动调用close()
        with open('processed_data.txt', 'w', encoding='utf-8') as output_file:
            output_file.write(f"处理后的数据：{data.upper()}")
            print("处理结果写入成功")
            
    except FileNotFoundError:
        print("文件不存在")
    except PermissionError:
        print("没有文件访问权限")
    except Exception as e:
        print(f"发生错误：{e}")

# 示例11：完整的文件操作异常处理
def robust_file_operations(filename, operation='read', content=None):
    """
    健壮的文件操作函数
    """
    modes = {
        'read': 'r',
        'write': 'w', 
        'append': 'a',
        'read_binary': 'rb',
        'write_binary': 'wb'
    }
    
    try:
        mode = modes.get(operation, 'r')
        encoding = None if 'b' in mode else 'utf-8'
        
        with open(filename, mode, encoding=encoding) as file:
            if operation == 'read':
                return file.read()
            elif operation in ['write', 'append'] and content:
                file.write(content)
                return True
                
    except FileNotFoundError:
        print(f"错误：文件 {filename} 不存在")
        return None
    except PermissionError:
        print(f"错误：没有权限访问文件 {filename}")
        return None
    except UnicodeDecodeError:
        print(f"错误：文件 {filename} 编码问题")
        return None
    except Exception as e:
        print(f"未知错误：{e}")
        return None

# 示例12：二进制文件复制
def copy_binary_file(source_path, target_path):
    """复制二进制文件（如图片、视频等）"""
    try:
        with open(source_path, 'rb') as source_file:
            with open(target_path, 'wb') as target_file:
                # 分块读取和写入，适合大文件
                chunk_size = 8192  # 8KB
                while True:
                    chunk = source_file.read(chunk_size)
                    if not chunk:
                        break
                    target_file.write(chunk)
        print(f"文件复制成功：{source_path} -> {target_path}")
    except Exception as e:
        print(f"文件复制失败：{e}")
# 示例13：处理图片文件
def process_image_file():
    """处理图片文件的示例"""
    try:
        with open('image.jpg', 'rb') as img_file:
            image_data = img_file.read()
            print(f"图片文件大小：{len(image_data)} 字节")
            # 这里可以添加图片处理逻辑
            # 例如：修改图片、添加水印等
    except FileNotFoundError:
        print("图片文件不存在")

import csv

# 示例14：CSV文件读取
def read_csv_file():
    """读取CSV文件"""
    with open('data.csv', 'r', encoding='utf-8', newline='') as csvfile:
        reader = csv.reader(csvfile)
        header = next(reader)  # 读取表头
        print(f"表头：{header}")
        
        for row_num, row in enumerate(reader, 1):
            print(f"第{row_num}行：{row}")

# 示例15：CSV文件写入
def write_csv_file():
    """写入CSV文件"""
    data = [
        ['姓名', '年龄', '城市'],
        ['张三', '25', '北京'],
        ['李四', '30', '上海'],
        ['王五', '28', '广州']
    ]
    
    with open('output.csv', 'w', encoding='utf-8', newline='') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerows(data)
    print("CSV文件写入完成")

import json

# 示例16：JSON文件读写
def json_file_operations():
    """JSON文件的读取和写入"""
    
    # 写入JSON文件
    data = {
        "name": "Python文件操作",
        "version": "1.0",
        "features": ["读取", "写入", "追加", "二进制处理"],
        "author": {
            "name": "Developer",
            "email": "dev@example.com"
        }
    }
    
    # 写入JSON文件
    with open('config.json', 'w', encoding='utf-8') as json_file:
        json.dump(data, json_file, indent=2, ensure_ascii=False)
    print("JSON文件写入完成")
    
    # 读取JSON文件
    with open('config.json', 'r', encoding='utf-8') as json_file:
        loaded_data = json.load(json_file)
        print("JSON文件内容：")
        print(json.dumps(loaded_data, indent=2, ensure_ascii=False))

# 示例17：使用seek方法移动文件指针
def file_pointer_operations():
    """文件指针操作示例"""
    with open('example.txt', 'r+', encoding='utf-8') as file:
        # 读取前10个字符
        start_content = file.read(10)
        print(f"前10个字符：{start_content}")
        
        # 移动到文件末尾
        file.seek(0, 2)  # 第二个参数：0=开头，1=当前位置，2=末尾
        file.write("
这是追加的内容")
        
        # 回到文件开头
        file.seek(0)
        full_content = file.read()
        print("完整文件内容：")
        print(full_content)

# 示例18：处理不同编码的文件
def handle_different_encodings():
    """处理不同编码格式的文件"""
    encodings = ['utf-8', 'gbk', 'gb2312', 'latin-1']
    filename = 'encoded_file.txt'
    
    for encoding in encodings:
        try:
            with open(filename, 'r', encoding=encoding) as file:
                content = file.read()
                print(f"使用 {encoding} 编码成功读取文件")
                break
        except UnicodeDecodeError:
            print(f"使用 {encoding} 编码读取失败，尝试下一种编码")
            continue
    else:
        print("所有编码尝试都失败了")

# 示例19：日志文件分析
def analyze_log_file(log_file_path):
    """分析日志文件"""
    error_count = 0
    warning_count = 0
    info_count = 0
    
    with open(log_file_path, 'r', encoding='utf-8') as log_file:
        for line_num, line in enumerate(log_file, 1):
            line = line.strip().lower()
            
            if 'error' in line:
                error_count += 1
                print(f"第{line_num}行发现错误：{line}")
            elif 'warning' in line:
                warning_count += 1
            elif 'info' in line:
                info_count += 1
    
    print(f"
日志分析结果：")
    print(f"错误数量：{error_count}")
    print(f"警告数量：{warning_count}")
    print(f"信息数量：{info_count}")
    print(f"总行数：{line_num}")

# 示例20：配置文件读写
class ConfigManager:
    """配置文件管理器"""
    
    def __init__(self, config_file='config.ini'):
        self.config_file = config_file
        self.config = {}
    
    def load_config(self):
        """加载配置文件"""
        try:
            with open(self.config_file, 'r', encoding='utf-8') as file:
                for line in file:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        key, value = line.split('=', 1)
                        self.config[key.strip()] = value.strip()
            print("配置文件加载成功")
        except FileNotFoundError:
            print("配置文件不存在，使用默认配置")
    
    def save_config(self):
        """保存配置文件"""
        with open(self.config_file, 'w', encoding='utf-8') as file:
            file.write("# 应用程序配置
")
            for key, value in self.config.items():
                file.write(f"{key}={value}
")
        print("配置文件保存成功")
    
    def get_value(self, key, default=None):
        """获取配置值"""
        return self.config.get(key, default)
    
    def set_value(self, key, value):
        """设置配置值"""
        self.config[key] = value

# 使用示例
config_mgr = ConfigManager()
config_mgr.load_config()
config_mgr.set_value('database_host', 'localhost')
config_mgr.set_value('database_port', '5432')
config_mgr.save_config()

# 示例21：高效文件处理模式
def efficient_file_processing(input_file, output_file):
    """高效的文件处理模式"""
    
    # 使用生成器表达式逐行处理，节省内存
    with open(input_file, 'r', encoding='utf-8') as infile, \
         open(output_file, 'w', encoding='utf-8') as outfile:
        
        # 逐行处理，避免一次性加载整个文件到内存
        processed_lines = (process_line(line) for line in infile)
        
        for processed_line in processed_lines:
            outfile.write(processed_line)

def process_line(line):
    """处理单行数据的示例函数"""
    return line.upper()  # 这里可以替换为实际的处理逻辑