第9：文件操作与异常处理

目录

• 文件操作基础
• 文件打开与关闭
• 文件读取
• 文件写入
• 文件定位
• 目录操作
• 异常处理基础
• 常见异常类型
• 自定义异常
• 上下文管理器
• 日志记录
• 实际应用示例

文件操作基础

文件操作是程序与外部存储交互的重要方式，Python提供了丰富的文件操作功能。

文件操作模式

# 文件打开模式
# 'r'  - 只读模式（默认）
# 'w'  - 写入模式（覆盖原文件）
# 'a'  - 追加模式
# 'x'  - 独占创建模式
# 'b'  - 二进制模式
# 't'  - 文本模式（默认）
# '+'  - 读写模式

# 常见组合
# 'rt' 或 'r'  - 文本只读
# 'wt' 或 'w'  - 文本写入
# 'rb'         - 二进制只读
# 'wb'         - 二进制写入
# 'r+'         - 文本读写
# 'w+'         - 文本读写（覆盖）
# 'a+'         - 文本读写（追加）

文件打开与关闭

正确打开和关闭文件是文件操作的基础。

基本文件操作

# 打开文件
file = open("example.txt", "r", encoding="utf-8")

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

# 关闭文件
file.close()

# 检查文件是否关闭
print(file.closed)  # 输出：True

文件对象属性

file = open("example.txt", "r", encoding="utf-8")

print(file.name)      # 文件名
print(file.mode)      # 打开模式
print(file.encoding)  # 编码方式

file.close()

文件读取

Python提供了多种读取文件内容的方法。

read() 方法

# 读取整个文件
with open("example.txt", "r", encoding="utf-8") as file:
    content = file.read()
    print(content)

# 读取指定字符数
with open("example.txt", "r", encoding="utf-8") as file:
    chunk = file.read(10)  # 读取前10个字符
    print(chunk)

readline() 方法

# 逐行读取
with open("example.txt", "r", encoding="utf-8") as file:
    line1 = file.readline()  # 读取第一行
    line2 = file.readline()  # 读取第二行
    print(f"第一行：{line1}")
    print(f"第二行：{line2}")

# 读取指定字符数的一行
with open("example.txt", "r", encoding="utf-8") as file:
    partial_line = file.readline(5)  # 读取一行的前5个字符
    print(partial_line)

readlines() 方法

# 读取所有行到列表
with open("example.txt", "r", encoding="utf-8") as file:
    lines = file.readlines()
    for i, line in enumerate(lines, 1):
        print(f"第{i}行：{line.strip()}")  # strip()去除换行符

迭代器方式读取

# 最推荐的方式：使用迭代器逐行读取
with open("example.txt", "r", encoding="utf-8") as file:
    for line_num, line in enumerate(file, 1):
        print(f"第{line_num}行：{line.strip()}")

文件写入

文件写入操作允许我们创建和修改文件内容。

write() 方法

# 写入文本
with open("output.txt", "w", encoding="utf-8") as file:
    file.write("Hello, World!\n")
    file.write("这是第二行内容\n")

# 写入多行
lines = ["第一行\n", "第二行\n", "第三行\n"]
with open("output.txt", "w", encoding="utf-8") as file:
    file.writelines(lines)

追加模式

# 追加内容到文件末尾
with open("output.txt", "a", encoding="utf-8") as file:
    file.write("这是追加的内容\n")

格式化写入

# 格式化写入数据
data = [
    {"name": "张三", "age": 25, "city": "北京"},
    {"name": "李四", "age": 30, "city": "上海"},
    {"name": "王五", "age": 35, "city": "广州"}
]

with open("users.txt", "w", encoding="utf-8") as file:
    file.write("姓名\t年龄\t城市\n")
    file.write("-" * 20 + "\n")
    for user in data:
        file.write(f"{user['name']}\t{user['age']}\t{user['city']}\n")

文件定位

文件定位允许我们在文件中移动读写位置。

seek() 和 tell() 方法

with open("example.txt", "r", encoding="utf-8") as file:
    # 获取当前位置
    print(f"当前位置：{file.tell()}")  # 输出：0
    
    # 读取一些内容
    content = file.read(10)
    print(f"读取内容：{content}")
    print(f"当前位置：{file.tell()}")  # 输出：10
    
    # 移动到文件开头
    file.seek(0)
    print(f"当前位置：{file.tell()}")  # 输出：0
    
    # 移动到指定位置
    file.seek(5)
    content = file.read(5)
    print(f"从位置5读取：{content}")

文件指针操作

# 相对定位
with open("example.txt", "r", encoding="utf-8") as file:
    file.seek(10)           # 绝对定位到位置10
    file.seek(5, 0)         # 从文件开头移动5个字节
    file.seek(5, 1)         # 从当前位置移动5个字节
    file.seek(-5, 2)        # 从文件末尾倒数5个字节

目录操作

Python的os和pathlib模块提供了目录操作功能。

os模块目录操作

import os

# 获取当前工作目录
current_dir = os.getcwd()
print(f"当前目录：{current_dir}")

# 切换目录
# os.chdir("/path/to/directory")

# 列出目录内容
files = os.listdir(".")
print(f"当前目录文件：{files}")

# 创建目录
# os.mkdir("new_directory")        # 创建单级目录
# os.makedirs("parent/child")      # 创建多级目录

# 删除目录
# os.rmdir("empty_directory")      # 删除空目录
# os.removedirs("parent/child")    # 删除多级空目录

# 检查路径是否存在
if os.path.exists("example.txt"):
    print("文件存在")

# 检查是否为文件或目录
if os.path.isfile("example.txt"):
    print("这是一个文件")

if os.path.isdir("documents"):
    print("这是一个目录")

pathlib模块（推荐）

from pathlib import Path

# 创建路径对象
path = Path("example.txt")

# 检查文件是否存在
if path.exists():
    print("文件存在")

# 检查是否为文件
if path.is_file():
    print("这是一个文件")

# 检查是否为目录
if path.is_dir():
    print("这是一个目录")

# 获取文件信息
if path.exists():
    stat = path.stat()
    print(f"文件大小：{stat.st_size} 字节")
    print(f"修改时间：{stat.st_mtime}")

# 列出目录内容
current_path = Path(".")
for item in current_path.iterdir():
    if item.is_file():
        print(f"文件：{item}")
    elif item.is_dir():
        print(f"目录：{item}")

# 创建目录
# new_dir = Path("new_directory")
# new_dir.mkdir(exist_ok=True)  # exist_ok=True避免目录已存在时的错误

# 路径操作
data_path = Path("data") / "users" / "profiles.txt"
print(f"完整路径：{data_path}")

异常处理基础

异常处理允许程序优雅地处理运行时错误。

try-except语句

# 基本异常处理
try:
    result = 10 / 0
except ZeroDivisionError:
    print("不能除以零")

# 捕获多种异常
try:
    number = int(input("请输入一个数字："))
    result = 10 / number
    print(f"结果：{result}")
except ValueError:
    print("请输入有效的数字")
except ZeroDivisionError:
    print("不能除以零")
except Exception as e:
    print(f"发生了未知错误：{e}")

else和finally子句

# else子句：没有异常时执行
# finally子句：无论是否有异常都会执行
try:
    file = open("example.txt", "r", encoding="utf-8")
    content = file.read()
except FileNotFoundError:
    print("文件未找到")
except Exception as e:
    print(f"读取文件时发生错误：{e}")
else:
    print("文件读取成功")
    print(f"文件内容：{content}")
finally:
    try:
        file.close()
        print("文件已关闭")
    except:
        pass

异常信息获取

try:
    result = 10 / 0
except ZeroDivisionError as e:
    print(f"异常类型：{type(e).__name__}")
    print(f"异常信息：{e}")
    print(f"异常参数：{e.args}")

常见异常类型

Python提供了多种内置异常类型。

内置异常类型

# 常见异常类型示例
exceptions_examples = [
    # ValueError - 值错误
    ("int('abc')", ValueError),
    
    # TypeError - 类型错误
    ("'hello' + 5", TypeError),
    
    # IndexError - 索引错误
    ("[1, 2, 3][5]", IndexError),
    
    # KeyError - 键错误
    ("{'a': 1}['b']", KeyError),
    
    # FileNotFoundError - 文件未找到
    ("open('nonexistent.txt')", FileNotFoundError),
    
    # ZeroDivisionError - 除零错误
    ("10 / 0", ZeroDivisionError),
    
    # AttributeError - 属性错误
    ("'hello'.nonexistent_method()", AttributeError)
]

# 演示异常处理
def demonstrate_exceptions():
    try:
        # 触发ValueError
        int("abc")
    except ValueError as e:
        print(f"ValueError: {e}")
    
    try:
        # 触发IndexError
        [1, 2, 3][5]
    except IndexError as e:
        print(f"IndexError: {e}")
    
    try:
        # 触发KeyError
        {"a": 1}["b"]
    except KeyError as e:
        print(f"KeyError: {e}")

demonstrate_exceptions()

自定义异常

我们可以创建自定义异常类型来处理特定的错误情况。

创建自定义异常

# 定义自定义异常类
class CustomError(Exception):
    """自定义异常基类"""
    pass

class ValidationError(CustomError):
    """验证错误"""
    def __init__(self, message, error_code=None):
        super().__init__(message)
        self.error_code = error_code

class NetworkError(CustomError):
    """网络错误"""
    pass

# 使用自定义异常
def validate_age(age):
    """验证年龄"""
    if age < 0:
        raise ValidationError("年龄不能为负数", error_code="NEGATIVE_AGE")
    elif age > 150:
        raise ValidationError("年龄不能超过150岁", error_code="INVALID_AGE")
    return True

# 处理自定义异常
try:
    validate_age(-5)
except ValidationError as e:
    print(f"验证失败：{e}")
    if hasattr(e, 'error_code'):
        print(f"错误代码：{e.error_code}")

上下文管理器

上下文管理器确保资源得到正确管理，即使发生异常。

with语句

# 文件操作的上下文管理
with open("example.txt", "r", encoding="utf-8") as file:
    content = file.read()
    # 文件会自动关闭，即使发生异常

# 等价的传统方式
file = open("example.txt", "r", encoding="utf-8")
try:
    content = file.read()
finally:
    file.close()  # 确保文件被关闭

自定义上下文管理器

# 使用类实现上下文管理器
class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        print("打开文件")
        self.file = open(self.filename, self.mode, encoding="utf-8")
        return self.file
    
    def __exit__(self, exc_type, exc_value, exc_traceback):
        print("关闭文件")
        if self.file:
            self.file.close()
        # 返回False表示不抑制异常
        return False

# 使用自定义上下文管理器
with FileManager("example.txt", "r") as file:
    content = file.read()
    print(content)

# 使用contextlib装饰器
from contextlib import contextmanager

@contextmanager
def database_connection():
    print("连接数据库")
    try:
        # 模拟数据库连接
        connection = "database_connection"
        yield connection
    finally:
        print("关闭数据库连接")

# 使用contextmanager
with database_connection() as conn:
    print(f"使用连接：{conn}")

日志记录

日志记录是跟踪程序运行状态的重要工具。

logging模块基础

import logging

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

# 创建日志记录器
logger = logging.getLogger(__name__)

# 不同级别的日志
logger.debug("这是调试信息")
logger.info("这是普通信息")
logger.warning("这是警告信息")
logger.error("这是错误信息")
logger.critical("这是严重错误信息")

日志处理

import logging

# 创建日志记录器
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 创建文件处理器
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setLevel(logging.INFO)

# 创建控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# 创建格式化器
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

# 添加处理器到记录器
logger.addHandler(file_handler)
logger.addHandler(console_handler)

# 使用日志
logger.debug("调试信息")
logger.info("应用程序启动")
logger.warning("这是一个警告")
logger.error("发生错误")

实际应用示例

import os
import json
import csv
import logging
from pathlib import Path

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 文件备份工具
class FileBackup:
    """文件备份工具"""
    
    def __init__(self, source_dir, backup_dir):
        self.source_dir = Path(source_dir)
        self.backup_dir = Path(backup_dir)
        self.backup_dir.mkdir(exist_ok=True)
    
    def backup_files(self):
        """备份文件"""
        try:
            if not self.source_dir.exists():
                raise FileNotFoundError(f"源目录不存在：{self.source_dir}")
            
            backed_up = 0
            for file_path in self.source_dir.rglob("*"):
                if file_path.is_file():
                    self._backup_file(file_path)
                    backed_up += 1
            
            logger.info(f"成功备份 {backed_up} 个文件")
            return backed_up
            
        except Exception as e:
            logger.error(f"备份过程中发生错误：{e}")
            raise
    
    def _backup_file(self, file_path):
        """备份单个文件"""
        try:
            # 计算相对路径
            relative_path = file_path.relative_to(self.source_dir)
            backup_path = self.backup_dir / relative_path
            
            # 创建目录
            backup_path.parent.mkdir(parents=True, exist_ok=True)
            
            # 复制文件
            with open(file_path, 'rb') as src, open(backup_path, 'wb') as dst:
                dst.write(src.read())
            
            logger.debug(f"已备份：{file_path} -> {backup_path}")
            
        except Exception as e:
            logger.error(f"备份文件 {file_path} 时发生错误：{e}")
            raise

# 配置文件管理器
class ConfigManager:
    """配置文件管理器"""
    
    def __init__(self, config_file="config.json"):
        self.config_file = Path(config_file)
        self.config = {}
        self.load_config()
    
    def load_config(self):
        """加载配置"""
        try:
            if self.config_file.exists():
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    self.config = json.load(f)
                logger.info("配置文件加载成功")
            else:
                logger.warning("配置文件不存在，使用默认配置")
                self.config = self._get_default_config()
        except json.JSONDecodeError as e:
            logger.error(f"配置文件格式错误：{e}")
            self.config = self._get_default_config()
        except Exception as e:
            logger.error(f"加载配置文件时发生错误：{e}")
            self.config = self._get_default_config()
    
    def save_config(self):
        """保存配置"""
        try:
            with open(self.config_file, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, ensure_ascii=False, indent=2)
            logger.info("配置文件保存成功")
        except Exception as e:
            logger.error(f"保存配置文件时发生错误：{e}")
            raise
    
    def get(self, key, default=None):
        """获取配置值"""
        return self.config.get(key, default)
    
    def set(self, key, value):
        """设置配置值"""
        self.config[key] = value
    
    def _get_default_config(self):
        """获取默认配置"""
        return {
            "app_name": "MyApp",
            "version": "1.0.0",
            "debug": False,
            "max_connections": 100
        }

# CSV数据处理器
class CSVProcessor:
    """CSV数据处理器"""
    
    def __init__(self, csv_file):
        self.csv_file = Path(csv_file)
    
    def read_csv(self):
        """读取CSV文件"""
        try:
            data = []
            with open(self.csv_file, 'r', encoding='utf-8') as f:
                reader = csv.DictReader(f)
                for row in reader:
                    data.append(row)
            return data
        except FileNotFoundError:
            logger.error(f"CSV文件不存在：{self.csv_file}")
            raise
        except Exception as e:
            logger.error(f"读取CSV文件时发生错误：{e}")
            raise
    
    def write_csv(self, data, fieldnames):
        """写入CSV文件"""
        try:
            with open(self.csv_file, 'w', newline='', encoding='utf-8') as f:
                writer = csv.DictWriter(f, fieldnames=fieldnames)
                writer.writeheader()
                writer.writerows(data)
            logger.info(f"CSV文件写入成功：{self.csv_file}")
        except Exception as e:
            logger.error(f"写入CSV文件时发生错误：{e}")
            raise
    
    def filter_data(self, data, condition):
        """过滤数据"""
        return [row for row in data if condition(row)]

# 使用示例
def main():
    """主函数示例"""
    try:
        # 配置管理
        config = ConfigManager("app_config.json")
        config.set("debug", True)
        config.save_config()
        
        print(f"应用名称：{config.get('app_name')}")
        print(f"调试模式：{config.get('debug')}")
        
        # CSV处理
        sample_data = [
            {"name": "张三", "age": "25", "city": "北京"},
            {"name": "李四", "age": "30", "city": "上海"},
            {"name": "王五", "age": "35", "city": "广州"}
        ]
        
        csv_processor = CSVProcessor("users.csv")
        csv_processor.write_csv(sample_data, ["name", "age", "city"])
        
        # 读取并过滤数据
        data = csv_processor.read_csv()
        young_people = csv_processor.filter_data(
            data, 
            lambda row: int(row["age"]) < 30
        )
        
        print("年龄小于30的用户：")
        for person in young_people:
            print(f"  {person['name']} - {person['age']}岁")
        
        # 文件备份
        backup = FileBackup(".", "backup")
        count = backup.backup_files()
        print(f"备份了 {count} 个文件")
        
    except Exception as e:
        logger.error(f"程序执行过程中发生错误：{e}")
        raise

if __name__ == "__main__":
    main()

总结

本篇教程详细介绍了Python中的文件操作和异常处理。我们学习了文件的读写操作、目录管理、异常处理机制、自定义异常、上下文管理器和日志记录等重要内容。

---