Python文件操作与异常处理（六）¶

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

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

# 关闭文件
file.close()

# 方法1：使用with语句（推荐）
with open("example.txt", "r", encoding="utf-8") as f:
    content = f.read()  # 读取全部内容

# 方法2：逐行读取
with open("example.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())  # strip()去除换行符

# 方法3：读取所有行到列表
with open("example.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()  # 返回列表

# 方法4：逐行读取（内存友好）
with open("example.txt", "r", encoding="utf-8") as f:
    while True:
        line = f.readline()
        if not line:
            break
        print(line.strip())

# 写入字符串
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")

# 写入多行
lines = ["行1\n", "行2\n", "行3\n"]
with open("output.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)

# 追加内容
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("追加的内容\n")

with open("example.txt", "r", encoding="utf-8") as f:
    print(f.tell())      # 0（当前位置）
    content1 = f.read(10)  # 读取前10个字符
    print(f.tell())      # 10
    f.seek(0)            # 回到文件开头
    print(f.tell())      # 0

def copy_file(source, destination):
    """复制文件"""
    try:
        with open(source, "r", encoding="utf-8") as src:
            content = src.read()

        with open(destination, "w", encoding="utf-8") as dst:
            dst.write(content)

        print(f"文件复制成功：{source} -> {destination}")
    except FileNotFoundError:
        print(f"源文件不存在：{source}")
    except Exception as e:
        print(f"复制失败：{e}")

# 使用
copy_file("input.txt", "output.txt")

def count_file_stats(filename):
    """统计文件的行数、单词数、字符数"""
    with open(filename, "r", encoding="utf-8") as f:
        content = f.read()

    lines = content.splitlines()
    words = content.split()
    chars = len(content)

    return {
        "行数": len(lines),
        "单词数": len(words),
        "字符数": chars
    }

# 使用
stats = count_file_stats("example.txt")
print(f"行数：{stats['行数']}")
print(f"单词数：{stats['单词数']}")
print(f"字符数：{stats['字符数']}")

# 基本语法
try:
    # 可能出错的代码
    result = 10 / 0
except ZeroDivisionError:
    # 处理特定异常
    print("除数不能为零！")

# 捕获多种异常
try:
    x = int(input("输入数字："))
    result = 10 / x
except ValueError:
    print("输入的不是数字！")
except ZeroDivisionError:
    print("除数不能为零！")

# 同时处理多种异常
try:
    x = int(input("输入数字："))
    result = 10 / x
except (ValueError, ZeroDivisionError) as e:
    print(f"发生错误：{e}")

# 捕获所有异常
try:
    risky_code()
except Exception as e:
    print(f"发生错误：{e}")

try:
    # 可能出错的代码
    f = open("example.txt", "r")
    content = f.read()
except FileNotFoundError:
    print("文件不存在！")
except PermissionError:
    print("没有权限读取文件！")
else:
    # 没有异常时执行
    print("文件读取成功！")
finally:
    # 无论是否有异常都执行
    print("执行清理操作...")

class InsufficientFundsError(Exception):
    """余额不足异常"""

    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        super().__init__(f"余额不足：当前{balance}元，需要{amount}元")

class BankAccount:
    def __init__(self, balance=0):
        self.balance = balance

    def withdraw(self, amount):
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)
        self.balance -= amount
        return self.balance

# 使用
account = BankAccount(100)
try:
    account.withdraw(150)
except InsufficientFundsError as e:
    print(e)  # 余额不足：当前100元，需要150元

class FileManager:
    """文件管理器上下文管理器"""

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

    def __enter__(self):
        """进入with语句时调用"""
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        """退出with语句时调用"""
        if self.file:
            self.file.close()
        # 返回False表示不抑制异常
        return False

# 使用
with FileManager("example.txt", "r") as f:
    content = f.read()

from contextlib import contextmanager

@contextmanager
def managed_resource(name):
    """上下文管理器装饰器"""
    print(f"获取资源：{name}")
    try:
        yield name  # yield的值作为as的值
    except Exception as e:
        print(f"处理异常：{e}")
    finally:
        print(f"释放资源：{name}")

# 使用
with managed_resource("数据库连接") as resource:
    print(f"使用资源：{resource}")
    # 可以在这里处理业务逻辑

# 输出：
# 获取资源：数据库连接
# 使用资源：数据库连接
# 释放资源：数据库连接

import datetime

class Logger:
    """简单日志记录器"""

    def __init__(self, filename):
        self.filename = filename

    def log(self, level, message):
        timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"[{timestamp}] [{level}] {message}\n"

        with open(self.filename, "a", encoding="utf-8") as f:
            f.write(log_entry)

    def info(self, message):
        self.log("INFO", message)

    def error(self, message):
        self.log("ERROR", message)

    def warning(self, message):
        self.log("WARNING", message)

# 使用
logger = Logger("app.log")
logger.info("程序启动")
logger.warning("内存使用率高")
logger.error("数据库连接失败")

import csv

def write_csv(filename, headers, data):
    """写入CSV文件"""
    with open(filename, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(headers)
        writer.writerows(data)

def read_csv(filename):
    """读取CSV文件"""
    with open(filename, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        return list(reader)

# 使用
headers = ["姓名", "年龄", "城市"]
data = [
    ["张三", 25, "北京"],
    ["李四", 30, "上海"],
    ["王五", 28, "广州"]
]

write_csv("students.csv", headers, data)
students = read_csv("students.csv")
for stu in students:
    print(stu)

import json

def load_config(filename):
    """加载JSON配置文件"""
    try:
        with open(filename, "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        print(f"配置文件不存在：{filename}")
        return {}
    except json.JSONDecodeError as e:
        print(f"JSON解析错误：{e}")
        return {}

def save_config(filename, config):
    """保存配置到JSON文件"""
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(config, f, indent=4, ensure_ascii=False)

# 使用
config = {
    "database": {
        "host": "localhost",
        "port": 3306,
        "name": "mydb"
    },
    "debug": True
}

save_config("config.json", config)
loaded_config = load_config("config.json")
print(loaded_config)