异常处理机制

# 错误码风格（C 语言思维方式）—— 容易遗漏检查
def divide_error_code(a, b):
    if b == 0:
        return None, "除数不能为零"   # 返回值混用了结果和错误
    return a / b, None

result, err = divide_error_code(10, 0)
# 如果忘记检查 err，直接使用 result 就会产生 NoneType 错误
print(result + 1)  # TypeError: unsupported operand type(s) for +: 'NoneType' and 'int'

# 异常风格（Pythonic 方式）—— 错误无法被忽略
def divide_exception(a, b):
    if b == 0:
        raise ZeroDivisionError("除数不能为零")  # 错误通过异常通道传递
    return a / b

try:
    result = divide_exception(10, 0)   # 正常返回值只包含业务结果
    print(result + 1)                  # 只有成功时才会执行到这里
except ZeroDivisionError as e:
    print(f"错误：{e}")                 # 错误处理与业务逻辑清晰分离

# 继承关系决定捕获行为
try:
    raise FileNotFoundError("配置文件不存在")
except OSError:           # FileNotFoundError 是 OSError 的子类，会被捕获
    print("被 OSError 捕获")

# 捕获顺序错误示例
try:
    raise ValueError("值错误")
except Exception:         # 先匹配，捕获了所有 Exception 子类
    print("被 Exception 捕获")
except ValueError:        # 永远不会执行！ValueError 已经被上面的 Exception 捕获
    print("永远不会到达这里")

# KeyboardInterrupt 不是 Exception 的子类
try:
    raise KeyboardInterrupt
except Exception:         # 不会捕获 KeyboardInterrupt！
    print("不会被捕获")
except BaseException:     # 需要 BaseException 才能捕获
    print("被 BaseException 捕获")

BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
    ├── ArithmeticError
    │   ├── ZeroDivisionError
    │   ├── FloatingPointError
    │   └── OverflowError
    ├── LookupError
    │   ├── IndexError
    │   └── KeyError
    ├── OSError
    │   ├── FileNotFoundError
    │   ├── PermissionError
    │   └── ... (其他 OSError 子类)
    ├── ValueError
    ├── TypeError
    ├── ImportError
    ├── SyntaxError
    │   └── IndentationError
    │       └── TabError
    └── ... (更多异常类)

try:
    # 可能引发异常的代码块
    ...
except ExceptionType as e:
    # 处理特定类型的异常
    ...
except AnotherExceptionType as e:
    # 处理另一种类型的异常
    ...
else:
    # 如果没有异常发生，执行此代码块
    ...
finally:
    # 无论是否发生异常，都会执行此代码块
    ...

try:
    num_str = input("请输入一个非零整数: ")
    num = int(num_str)
    result = 100 / num
except ValueError:
    print(f"输入无效: '{num_str}' 不是一个有效的整数。")
except ZeroDivisionError:
    print("错误：除数不能为零。")

try:
    # ... 同样的代码 ...
    num_str = input("请输入一个非零整数: ")
    num = int(num_str)
    result = 100 / num
except (ValueError, ZeroDivisionError) as e:
    print(f"发生了一个输入或计算错误: {e}")

# 捕获多种异常的完整可运行示例
def safe_divide(a, b):
    """安全除法函数，演示多种异常捕获方式"""
    try:
        # 第一步：尝试将输入转换为浮点数
        x = float(a)                   # 可能抛出 ValueError
        y = float(b)                   # 可能抛出 ValueError

        # 第二步：执行除法运算
        result = x / y                 # 可能抛出 ZeroDivisionError

    except ValueError as e:
        # 捕获类型转换失败：输入不是有效数字
        print(f"[ValueError] 输入无效，无法转换为数字: {e}")

    except ZeroDivisionError as e:
        # 捕获除零错误：除数为 0
        print(f"[ZeroDivisionError] 除数不能为零: {e}")

    except (TypeError, NameError) as e:
        # 用元组同时捕获多种异常，统一处理
        print(f"[TypeError/NameError] 类型或名称错误: {e}")

    else:
        # 没有异常时执行：输出计算结果
        print(f"计算成功: {x} / {y} = {result}")

    finally:
        # 无论如何都会执行：收尾工作
        print("--- safe_divide 调用结束 ---\n")


# 测试各种情况
safe_divide(10, 3)          # 正常计算
safe_divide(10, 0)          # ZeroDivisionError
safe_divide("abc", 2)       # ValueError
safe_divide(10, "xyz")      # ValueError

# 异常对象的三大属性：类型、值、回溯
import sys
import traceback

def demonstrate_exception_attributes():
    """演示异常对象的属性"""
    try:
        # 故意触发一个异常
        numbers = [1, 2, 3]
        print(numbers[10])              # IndexError: 索引超出范围

    except IndexError as e:
        # 属性 1：异常的类型 —— 即异常的类对象
        exc_type = type(e)              # <class 'IndexError'>
        print(f"异常类型: {exc_type.__name__}")        # IndexError

        # 属性 2：异常的值 —— 即异常实例本身，包含错误消息
        print(f"异常值（str）: {e}")                    # list index out of range
        print(f"异常值（repr）: {repr(e)}")            # IndexError('list index out of range')

        # 属性 3：异常的回溯对象 —— 记录了异常发生时的调用栈
        exc_tb = e.__traceback__        # traceback 对象
        print(f"回溯对象: {exc_tb}")

        # 使用 sys.exc_info() 获取三元组 (type, value, traceback)
        exc_info = sys.exc_info()
        print(f"sys.exc_info() 类型: {exc_info[0]}")   # <class 'IndexError'>
        print(f"sys.exc_info() 值:   {exc_info[1]}")   # IndexError 实例
        print(f"sys.exc_info() 回溯: {exc_info[2]}")   # traceback 对象

        # 使用 traceback 模块格式化完整的调用栈
        print("\n--- 完整回溯信息 ---")
        traceback.print_exc()           # 打印格式化的 traceback

        # 异常的 args 属性：传递给异常构造函数的参数元组
        print(f"\n异常参数 (args): {e.args}")          # ('list index out of range',)

demonstrate_exception_attributes()

try:
    num_str = input("请输入一个非零整数: ")
    num = int(num_str)
    result = 100 / num
except (ValueError, ZeroDivisionError) as e:
    print(f"错误: {e}")
else:
    # 仅在 try 块成功时执行
    print(f"计算成功！结果是: {result:.2f}")

# try-except-else-finally 的完整可运行示例
def read_file_safely(filepath):
    """
    安全读取文件内容的完整示例
    演示 try-except-else-finally 的完整执行流程
    """
    file_handle = None                  # 初始化文件句柄，确保 finally 中可以安全引用

    try:
        # try 块：只放置"可能出错"的代码
        file_handle = open(filepath, 'r', encoding='utf-8')  # 可能抛出 FileNotFoundError
        content = file_handle.read()                          # 可能抛出 IOError

    except FileNotFoundError:
        # except 块：处理"文件不存在"的情况
        print(f"[except] 错误：文件 '{filepath}' 不存在")
        return None                     # 提前返回，但 finally 仍会执行

    except PermissionError:
        # except 块：处理"权限不足"的情况
        print(f"[except] 错误：没有权限读取 '{filepath}'")
        return None

    except UnicodeDecodeError as e:
        # except 块：处理"编码错误"的情况
        print(f"[except] 错误：文件编码无法解析 - {e}")
        return None

    else:
        # else 块：try 块没有任何异常时执行
        # 这里放"成功后"的逻辑，避免将成功逻辑混入 try 块
        print(f"[else] 文件读取成功，共 {len(content)} 个字符")
        return content                  # 提前返回，但 finally 仍会执行

    finally:
        # finally 块：无论是否异常、无论是否 return，都会执行
        # 用于释放资源，确保文件句柄被关闭
        if file_handle is not None:
            file_handle.close()
            print("[finally] 文件句柄已关闭")
        else:
            print("[finally] 无需关闭（文件未成功打开）")


# 测试：读取不存在的文件
print("=== 测试1：文件不存在 ===")
result1 = read_file_safely("nonexistent_file.txt")

# 测试：读取存在的文件（请先创建一个测试文件）
# print("\n=== 测试2：文件存在 ===")
# with open("test_file.txt", "w", encoding="utf-8") as f:
#     f.write("Hello, Python 异常处理！")
# result2 = read_file_safely("test_file.txt")

file = None
try:
    file = open("non_existent_file.txt", "r", encoding="utf-8")
    content = file.read()
    print(content)
except FileNotFoundError:
    print("错误：文件未找到。")
except Exception as e:
    print(f"发生未知错误: {e}")
finally:
    if file:
        file.close()
        print("文件已关闭，资源已释放。")

try:
    with open("non_existent_file.txt", "r", encoding="utf-8") as file:
        content = file.read()
        print(content)
except FileNotFoundError:
    print("错误：文件未找到。")
except Exception as e:
    print(f"发生未知错误: {e}")

# 当代码块执行完毕或发生异常跳出时，文件会自动关闭，无需手动调用 file.close()

try:
    result = 10 / 0
except:
    print("发生了错误。")

# 栈回溯过程的可视化演示
def inner():
    """最内层函数：异常在这里发生"""
    x = 10                             # 局部变量 x，栈回溯时会被销毁
    y = 0                              # 局部变量 y
    return x / y                       # ZeroDivisionError 在这里发生！

def middle():
    """中间层函数：调用 inner，没有捕获异常"""
    result = inner()                   # 异常从 inner 传播到这里
    print("这行不会执行")               # 因为异常已经传播，此行被跳过
    return result

def outer():
    """最外层函数：调用 middle，捕获异常"""
    try:
        middle()                       # 异常从 middle 传播到这里
        print("这行也不会执行")          # 因为异常已经传播，此行被跳过
    except ZeroDivisionError as e:
        # 异常在这里被捕获，栈回溯停止
        print(f"在 outer 中捕获异常: {e}")
        print("inner 中的局部变量 x、y 已经随着栈帧销毁而不可访问")

outer()
# 输出：
# 在 outer 中捕获异常: division by zero
# inner 中的局部变量 x、y 已经随着栈帧销毁而不可访问

def func_a():
    print("进入 func_a")
    func_b()
    print("退出 func_a")  # 这行不会被执行

def func_b():
    print("进入 func_b")
    raise ValueError("来自 func_b 的错误")
    print("退出 func_b")  # 这行不会被执行

try:
    func_a()
except ValueError as e:
    print(f"捕捉到异常: {e}")

进入 func_a
进入 func_b
捕捉到异常: 来自 func_b 的错误

try:
    try:
        result = 10 / 0
    except ZeroDivisionError as e:
        print(f"内部捕捉到异常: {e}")
        raise  # 重新抛出异常
except Exception as e:
    print(f"外部捕捉到异常: {e}")

内部捕捉到异常: division by zero
外部捕捉到异常: division by zero

def divide(a, b):
    if b == 0:
        raise ZeroDivisionError("除数不能为零。")
    return a / b

try:
    result = divide(10, 0)
except ZeroDivisionError as e:
    print(e)

# 输出：除数不能为零。

try:
    try:
        result = 10 / 0
    except ZeroDivisionError:
        print("记录日志：发生了除零错误")
        raise  # 重新抛出异常
except ZeroDivisionError as e:
    print(f"最终处理：{e}")

# raise 和 re-raise 的完整可运行示例
import logging

# 配置日志
logging.basicConfig(level=logging.WARNING, format='%(levelname)s: %(message)s')

def process_data(data):
    """处理数据，演示 raise 的各种用法"""
    if data is None:
        # raise 一个新的异常：主动抛出错误
        raise ValueError("数据不能为 None")

    if isinstance(data, str):
        # raise 一个带自定义消息的 TypeError
        raise TypeError(f"期望数字类型，收到了字符串: '{data}'")

    return data * 2


def safe_process(data):
    """安全处理数据，演示 re-raise 的用法"""
    try:
        result = process_data(data)

    except ValueError as e:
        # 记录日志后重新抛出 —— 让上层调用者决定如何处理
        logging.warning(f"process_data 抛出 ValueError: {e}")
        raise                           # re-raise：保留原始异常的完整 traceback

    except TypeError as e:
        # 捕获后抛出一个不同类型的异常
        logging.warning(f"类型错误，转换为 RuntimeError: {e}")
        raise RuntimeError(f"数据处理失败: {e}") from e  # 使用异常链保留原始原因

    return result


# 测试 1：正常数据
try:
    print(safe_process(5))              # 输出: 10
except Exception as e:
    print(f"捕获: {e}")

# 测试 2：None 值 —— 触发 ValueError，被 re-raise
try:
    safe_process(None)
except ValueError as e:
    print(f"上层捕获 ValueError: {e}")

# 测试 3：字符串 —— 触发 TypeError，转换为 RuntimeError
try:
    safe_process("hello")
except RuntimeError as e:
    print(f"上层捕获 RuntimeError: {e}")
    if e.__cause__:
        print(f"  原始原因: {type(e.__cause__).__name__}: {e.__cause__}")

class MyCustomError(Exception):
    def __init__(self, message):
        super().__init__(message)

def greet(name):
    if not name:
        raise MyCustomError("名字不能为空。")
    print(f"Hello, {name}!")

try:
    greet("")
except MyCustomError as e:
    print(e)

# 自定义异常类的完整可运行示例
# ---- 定义异常层级 ----

class AppError(Exception):
    """应用程序基础异常，所有自定义异常的父类"""
    pass


class DatabaseError(AppError):
    """数据库相关错误"""
    def __init__(self, query, message="数据库操作失败"):
        self.query = query              # 记录出错的 SQL 查询
        self.message = message
        super().__init__(self.message)

    def __str__(self):
        return f"{self.message} | SQL: {self.query}"


class ConnectionError(DatabaseError):
    """数据库连接错误（继承自 DatabaseError）"""
    def __init__(self, host, port, query=""):
        self.host = host                # 记录连接的主机
        self.port = port                # 记录连接的端口
        super().__init__(query, f"无法连接到 {host}:{port}")


class ValidationError(AppError):
    """数据验证错误"""
    def __init__(self, field, value, reason=""):
        self.field = field              # 出错的字段名
        self.value = value              # 出错的值
        self.reason = reason            # 出错原因
        super().__init__(f"字段 '{field}' 验证失败: {reason} (值: {value})")


# ---- 使用自定义异常 ----

def validate_user(username, age):
    """验证用户数据，演示自定义异常的使用"""
    if not username or len(username) < 2:
        # 抛出验证异常，携带字段和值信息
        raise ValidationError("username", username, "用户名至少2个字符")

    if not isinstance(age, int) or age < 0 or age > 150:
        raise ValidationError("age", age, "年龄必须是 0-150 的整数")

    return {"username": username, "age": age}


def connect_database(host, port):
    """模拟数据库连接，演示异常层级"""
    if port < 0 or port > 65535:
        raise ConnectionError(host, port)

    # 模拟连接失败
    if host == "unreachable.host":
        raise ConnectionError(host, port)

    return f"已连接到 {host}:{port}"


# ---- 测试 ----

# 测试 1：验证错误
try:
    validate_user("A", 25)              # 用户名太短
except ValidationError as e:
    print(f"验证失败: {e}")
    print(f"  出错字段: {e.field}")
    print(f"  出错值: {e.value}")

# 测试 2：数据库连接错误
try:
    connect_database("unreachable.host", 5432)
except ConnectionError as e:
    print(f"连接失败: {e}")
    print(f"  主机: {e.host}")
    print(f"  端口: {e.port}")

# 测试 3：利用异常层级，用父类捕获所有数据库错误
try:
    connect_database("unreachable.host", 5432)
except DatabaseError as e:              # ConnectionError 是 DatabaseError 的子类
    print(f"数据库错误（父类捕获）: {e}")

# 测试 4：利用最顶层父类捕获所有自定义异常
try:
    validate_user("", -1)
except AppError as e:                   # ValidationError 是 AppError 的子类
    print(f"应用错误（顶层捕获）: {e}")

class AgeError(Exception):
    """年龄错误异常"""
    def __init__(self, age, message="年龄必须在 0 到 120 之间。"):
        self.age = age
        self.message = message
        super().__init__(self.message)

def check_age(age):
    if age < 0 or age > 120:
        raise AgeError(age)
    return age

try:
    age = check_age(150)
except AgeError as e:
    print(f"错误: {e}")


# 输出：错误: 年龄必须在 0 到 120 之间。

raise NewException("更高层级的错误信息") from original_exception

import json

class ConfigError(Exception):
    """自定义配置错误，用于封装所有与配置相关的失败。"""
    pass

def load_config(path):
    """加载配置文件并验证。"""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            data = json.load(f)

        if 'host' not in data or 'port' not in data:
            raise KeyError("配置文件中缺少 'host' 或 'port' 项")

        return data

    except FileNotFoundError as e:
        # 将 FileNotFoundError 链接到新的 ConfigError
        raise ConfigError(f"配置文件 '{path}' 不存在") from e

    except json.JSONDecodeError as e:
        # 将 JSONDecodeError 链接到新的 ConfigError
        raise ConfigError(f"配置文件 '{path}' 格式错误") from e

    except KeyError as e:
        # 将 KeyError 链接到新的 ConfigError
        raise ConfigError("配置项不完整") from e

# --- 测试 --- #
try:
    config = load_config("config.json")
    print("配置加载成功:", config)
except ConfigError as e:
    print(f"配置失败: {e}")
    if e.__cause__:
        print(f"  根本原因: {type(e.__cause__).__name__}: {e.__cause__}")

配置失败: 配置文件 'config.json' 不存在
  根本原因: FileNotFoundError: [Errno 2] No such file or directory: 'config.json'

# 异常链的完整可运行示例
import json

class PaymentError(Exception):
    """支付错误基类"""
    pass

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

class InsufficientFundsError(PaymentError):
    """余额不足"""
    pass


def call_payment_api(card_number, amount):
    """模拟调用支付 API（底层函数）"""
    if not card_number.isdigit():
        # 底层抛出 ValueError，表示卡号格式无效
        raise ValueError(f"卡号格式无效: {card_number}")
    if amount > 10000:
        # 底层抛出 RuntimeError，表示金额超限
        raise RuntimeError("单笔交易金额超限")
    return {"status": "success", "transaction_id": "TXN_12345"}


def process_payment(card_number, amount):
    """处理支付（高层函数），将底层异常转换为领域异常"""
    try:
        result = call_payment_api(card_number, amount)
        return result

    except ValueError as e:
        # 将底层 ValueError 转换为 NetworkError，保留原始原因
        raise NetworkError(f"支付请求格式错误: {e}") from e

    except RuntimeError as e:
        # 将底层 RuntimeError 转换为 InsufficientFundsError
        raise InsufficientFundsError(f"支付被拒绝: {e}") from e


# 测试：卡号格式错误
try:
    process_payment("ABC123", 100)
except PaymentError as e:               # 用父类捕获所有支付错误
    print(f"支付失败: {e}")
    if e.__cause__:                     # 通过 __cause__ 访问原始异常
        print(f"  底层原因: {type(e.__cause__).__name__}: {e.__cause__}")

# 测试：金额超限
try:
    process_payment("1234567890", 50000)
except PaymentError as e:
    print(f"支付失败: {e}")
    if e.__cause__:
        print(f"  底层原因: {type(e.__cause__).__name__}: {e.__cause__}")

# 测试：使用 raise ... from None 抑制异常链
# 有时你想抛出新异常但不想暴露底层原因（如安全考虑）
try:
    try:
        raise RuntimeError("敏感的内部错误信息")
    except RuntimeError:
        # from None 会将 __cause__ 设为 None，隐藏原始异常
        raise PaymentError("支付处理失败，请稍后重试") from None
except PaymentError as e:
    print(f"支付失败: {e}")
    print(f"  __cause__: {e.__cause__}")  # None，底层异常被隐藏

# assert 的完整可运行示例

def calculate_average(numbers):
    """
    计算数值列表的平均值
    使用 assert 进行前置条件检查
    """
    # 断言：输入必须是列表且非空
    assert isinstance(numbers, list), f"期望 list，收到了 {type(numbers).__name__}"
    assert len(numbers) > 0, "列表不能为空"

    total = sum(numbers)
    average = total / len(numbers)

    # 断言：结果应该是一个有限数字
    assert isinstance(average, (int, float)), "平均值应该是数字"

    return average


# 正常使用
result = calculate_average([10, 20, 30])
print(f"平均值: {result}")               # 20.0

# 触发断言：空列表
try:
    calculate_average([])
except AssertionError as e:
    print(f"断言失败: {e}")               # 列表不能为空

# 触发断言：类型错误
try:
    calculate_average("not a list")
except AssertionError as e:
    print(f"断言失败: {e}")               # 期望 list，收到了 str


# 重要：assert 可以通过 python -O 命令禁用！
# 运行 python -O script.py 时，所有 assert 语句会被跳过
# 因此不要在 assert 中放有副作用的代码

# 错误示范：assert 中有副作用
# assert file.close(), "文件关闭失败"  # -O 模式下 close() 不会执行！

# 正确做法：先执行操作，再断言结果
# file.close()
# assert file.closed, "文件关闭失败"

class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name
        self.connection = None

    def __enter__(self):
        print(f"连接到数据库 {self.db_name}")
        self.connection = f"连接到 {self.db_name}"
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("关闭数据库连接")
        self.connection = None
        # 如果返回 True，异常会被抑制
        # 如果返回 False 或 None，异常会继续传播
        return False

# 使用自定义上下文管理器
try:
    with DatabaseConnection("mydb") as conn:
        print(f"使用连接: {conn}")
        # 模拟异常
        raise ValueError("数据库操作失败")
except ValueError as e:
    print(f"捕获异常: {e}")

# 上下文管理器与异常的完整可运行示例

class Timer:
    """计时器上下文管理器，自动测量代码块执行时间"""
    import time

    def __init__(self, name="代码块"):
        self.name = name                # 代码块名称，用于日志输出
        self.start_time = None          # 开始时间
        self.elapsed = None             # 经过时间

    def __enter__(self):
        """进入 with 块时调用：开始计时"""
        import time
        self.start_time = time.time()
        print(f"[{self.name}] 开始执行...")
        return self                     # 返回自身，可以用 as 接收

    def __exit__(self, exc_type, exc_val, exc_tb):
        """退出 with 块时调用：停止计时并报告"""
        import time
        self.elapsed = time.time() - self.start_time

        if exc_type is None:
            # 正常退出：无异常
            print(f"[{self.name}] 正常完成，耗时 {self.elapsed:.4f} 秒")
        else:
            # 异常退出：exc_type 是异常类型，exc_val 是异常值
            print(f"[{self.name}] 异常退出 ({exc_type.__name__}: {exc_val})，"
                  f"耗时 {self.elapsed:.4f} 秒")

        # 返回 False：异常继续传播
        # 返回 True：异常被抑制（不推荐，除非有充分理由）
        return False


# 测试 1：正常执行
with Timer("数据处理"):
    total = sum(range(1000000))

# 测试 2：异常执行
try:
    with Timer("风险操作"):
        result = 10 / 0                 # 触发异常
except ZeroDivisionError:
    print("外部捕获到异常")


# 使用 contextlib.contextmanager 装饰器简化
from contextlib import contextmanager

@contextmanager
def temp_directory(path):
    """临时目录上下文管理器（简化版）"""
    import os
    os.makedirs(path, exist_ok=True)    # 创建目录
    print(f"创建临时目录: {path}")
    try:
        yield path                      # yield 之前的代码相当于 __enter__
    finally:
        # yield 之后的代码相当于 __exit__，finally 确保一定执行
        print(f"清理临时目录: {path}")
        # 实际项目中这里可以执行 shutil.rmtree(path) 删除临时目录


# 测试 contextmanager 装饰器版本
with temp_directory("/tmp/my_temp_dir") as tmp:
    print(f"在临时目录中工作: {tmp}")

from contextlib import contextmanager

@contextmanager
def file_manager(filename, mode):
    try:
        file = open(filename, mode)
        yield file
    finally:
        file.close()

# 使用简化版的上下文管理器
with file_manager("example.txt", "w") as f:
    f.write("Hello, world!")
# 文件会自动关闭

# warnings 模块的完整可运行示例
import warnings

# ---- 基本用法 ----

# 发出一个默认警告（默认只显示一次）
warnings.warn("这是一个普通警告")

# 发出 DeprecationWarning（弃用警告，默认被过滤器隐藏）
warnings.warn("旧函数已弃用，请使用新函数", DeprecationWarning)

# 发出 UserWarning（用户可见的警告）
warnings.warn("配置项 'old_option' 已弃用，请使用 'new_option'", UserWarning)


# ---- 在函数中使用警告 ----

def legacy_function(data):
    """
    旧版函数，使用警告提示用户迁移
    警告不会中断程序执行
    """
    if isinstance(data, str):
        # 发出 DeprecationWarning，但通过 stacklevel 指向调用者
        warnings.warn(
            "字符串参数已弃用，请传入列表",
            DeprecationWarning,
            stacklevel=2                # stacklevel=2 让警告指向调用者而非本函数
        )
        data = [data]

    return data


# 调用旧函数，会看到弃用警告
result = legacy_function("hello")
print(f"结果: {result}")                 # 结果: ['hello']


# ---- 控制警告行为 ----

# 将特定警告转为异常
warnings.filterwarnings("error", category=DeprecationWarning)

try:
    warnings.warn("这个弃用警告会变成异常", DeprecationWarning)
except DeprecationWarning as e:
    print(f"警告被转为异常: {e}")

# 忽略特定警告
warnings.filterwarnings("ignore", category=UserWarning)
warnings.warn("这个用户警告会被忽略", UserWarning)   # 不会显示

# 重置所有过滤器
warnings.resetwarnings()


# ---- 自定义警告类 ----

class PerformanceWarning(UserWarning):
    """性能警告：建议优化"""
    pass

def process_large_data(data, chunk_size=100):
    """处理大数据，当数据量大时发出性能警告"""
    if len(data) > chunk_size * 10:
        warnings.warn(
            f"数据量 ({len(data)}) 远超块大小 ({chunk_size})，建议增大 chunk_size",
            PerformanceWarning,
            stacklevel=2
        )
    return len(data)

process_large_data(list(range(100000)), chunk_size=100)

# traceback 模块的完整可运行示例
import traceback
import sys


def deep_function():
    """最深层函数：异常在这里发生"""
    data = {"key": "value"}
    return data["nonexistent_key"]      # KeyError: 键不存在

def middle_function():
    """中间层函数"""
    return deep_function()

def top_function():
    """顶层函数"""
    return middle_function()


# ---- 1. traceback.print_exc()：打印当前异常的格式化回溯 ----
print("=== 1. print_exc() ===")
try:
    top_function()
except KeyError:
    traceback.print_exc()               # 打印格式化的 traceback 到 stderr


# ---- 2. traceback.format_exc()：获取回溯的字符串 ----
print("\n=== 2. format_exc() ===")
try:
    top_function()
except KeyError:
    tb_str = traceback.format_exc()     # 返回字符串而非打印
    print(f"回溯字符串长度: {len(tb_str)}")
    print(tb_str[:200] + "...")         # 只打印前 200 个字符


# ---- 3. traceback.extract_tb()：提取回溯帧信息 ----
print("\n=== 3. extract_tb() ===")
try:
    top_function()
except KeyError as e:
    exc_type, exc_value, exc_tb = sys.exc_info()

    # extract_tb 返回 FrameSummary 对象列表
    frames = traceback.extract_tb(exc_tb)
    for i, frame in enumerate(frames):
        print(f"  帧 {i}: {frame.filename}:{frame.lineno} in {frame.name}")
        print(f"         {frame.line}")


# ---- 4. traceback.print_stack()：打印当前调用栈（无需异常） ----
print("\n=== 4. print_stack() ===")
def show_current_stack():
    """打印当前的调用栈，即使没有异常"""
    print("当前调用栈:")
    traceback.print_stack()             # 打印从程序入口到当前位置的完整调用栈

show_current_stack()


# ---- 5. 将 traceback 写入日志文件 ----
print("\n=== 5. 写入日志 ===")
import logging

logging.basicConfig(
    level=logging.ERROR,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

try:
    top_function()
except KeyError as e:
    # 使用 logging.exception 自动记录完整 traceback
    logging.exception("处理数据时发生错误")

    # 或者手动将 traceback 字符串写入日志
    # tb_str = traceback.format_exc()
    # logging.error("处理数据时发生错误:\n%s", tb_str)

import traceback

try:
    result = 10 / 0
except ZeroDivisionError:
    traceback.print_exc()

Traceback (most recent call last):
  File "example.py", line 3, in <module>
    result = 10 / 0
ZeroDivisionError: division by zero

import logging

logging.basicConfig(level=logging.ERROR, filename='app.log',format='%(asctime)s - %(levelname)s - %(message)s')

try:
    result = 10 / 'a'
except TypeError as e:
    logging.error("类型错误发生: %s", e, exc_info=True)

ERROR - 类型错误发生: unsupported operand type(s) for /: 'int' and 'str'
Traceback (most recent call last):
  File "example.py", line 6, in <module>
    result = 10 / 'a'
TypeError: unsupported operand type(s) for /: 'int' and 'str'

import logging

logging.basicConfig(level=logging.DEBUG)

try:
    result = 10 / 2
    logging.info("除法成功，结果为 %s", result)
except ZeroDivisionError as e:
    logging.error("除零错误: %s", e)

try:
    with open("data.txt", "r") as file:
        content = file.read()
        print(content)
except FileNotFoundError:
    print("文件未找到，请检查文件路径。")
except PermissionError:
    print("没有权限读取该文件。")
except IOError as e:
    print(f"文件操作失败: {e}")

import requests

url = "https://api.example.com/data"

try:
    response = requests.get(url, timeout=5)
    response.raise_for_status()  # 检查响应状态码
    data = response.json()
    print(data)
except requests.exceptions.HTTPError as http_err:
    print(f"HTTP 错误发生: {http_err}")
except requests.exceptions.ConnectionError as conn_err:
    print(f"连接错误发生: {conn_err}")
except requests.exceptions.Timeout as timeout_err:
    print(f"请求超时: {timeout_err}")
except requests.exceptions.RequestException as req_err:
    print(f"请求异常: {req_err}")

import sqlite3

db_path = "example.db"

try:
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
    user = cursor.fetchone()
    if user:
        print(f"User found: {user}")
    else:
        print("User not found.")
except sqlite3.Error as e:
    print(f"Database error occurred: {e}")
finally:
    if conn:
        conn.close()

try:
    age = int(input("Please enter your age: "))
    if age < 0:
        raise ValueError("Age cannot be negative.")
    print(f"Your age is {age}.")
except ValueError as ve:
    print(f"Invalid input: {ve}")

Please enter your age: twenty
Invalid input: invalid literal for int() with base 10: 'twenty'

Please enter your age: -5
Invalid input: Age cannot be negative.

Please enter your age: 25
Your age is 25.

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/divide', methods=['GET'])
def divide():
    try:
        a = float(request.args.get('a'))
        b = float(request.args.get('b'))
        result = a / b
        return jsonify({"result": result})
    except ZeroDivisionError:
        return jsonify({"error": "除数不能为零。"}), 400
    except (TypeError, ValueError):
        return jsonify({"error": "参数必须是数字。"}), 400
    except Exception as e:
        # 记录未知异常
        app.logger.error(f"未知错误: {e}")
        return jsonify({"error": "内部服务器错误。"}), 500

if __name__ == '__main__':
    app.run(debug=True)

GET /divide?a=10&b=2
# 响应: {"result": 5.0}

GET /divide?a=10&b=0
# 响应: {"error": "除数不能为零。"}, 状态码: 400

GET /divide?a=abc&b=2
# 响应: {"error": "参数必须是数字。"}, 状态码: 400

import threading
import queue

def worker(q):
    try:
        # 模拟发生异常
        raise ValueError("子线程发生错误")
    except Exception as e:
        q.put(e)

q = queue.Queue()
thread = threading.Thread(target=worker, args=(q,))
thread.start()
thread.join()

if not q.empty():
    error = q.get()
    print(f"捕捉到子线程异常: {error}")


# 输出：捕捉到子线程异常: 子线程发生错误

from concurrent.futures import ThreadPoolExecutor

def worker():
    raise ValueError("子线程发生错误")

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(worker)
    try:
        future.result()
    except Exception as e:
        print(f"捕捉到子线程异常: {e}")

# 输出：捕捉到子线程异常: 子线程发生错误

import asyncio

async def async_worker():
    await asyncio.sleep(1)
    raise ValueError("异步函数发生错误")

async def main():
    try:
        await async_worker()
    except ValueError as e:
        print(f"捕捉到异步异常: {e}")

asyncio.run(main())

# 输出：捕捉到异步异常: 异步函数发生错误

import asyncio

async def task1():
    await asyncio.sleep(1)
    raise ValueError("任务1发生错误")

async def task2():
    await asyncio.sleep(2)
    return "任务2完成"

async def main():
    results = await asyncio.gather(task1(), task2(), return_exceptions=True)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务{i+1}捕捉到异常: {result}")
        else:
            print(f"任务{i+1}结果: {result}")

asyncio.run(main())

任务1捕捉到异常: 任务1发生错误
任务2结果: 任务2完成

# 使用条件判断（更高效）
def divide_with_check(a, b):
    if b == 0:
        return None
    return a / b

# 使用异常处理（较低效）
def divide_with_exception(a, b):
    try:
        return a / b
    except ZeroDivisionError:
        return None

# 反模式：bare except 吞掉所有异常
try:
    do_something()
except:                    # 危险！连 KeyboardInterrupt 都会被捕获
    pass                   # 异常被吞掉，问题完全被隐藏

# 正确做法：捕获特定异常
try:
    do_something()
except (ValueError, IOError) as e:   # 只捕获已知可恢复的异常
    logging.error(f"操作失败: {e}")    # 至少记录日志

# 如果确实需要捕获所有"普通"异常
try:
    do_something()
except Exception as e:     # 不包括 KeyboardInterrupt 和 SystemExit
    logging.exception("未预期的错误")

# assert：开发阶段的内部不变量检查
def factorial(n):
    assert isinstance(n, int) and n >= 0, "n 必须是非负整数"  # 开发时检查
    if n <= 1:
        return 1
    return n * factorial(n - 1)

# raise：运行时的输入验证（即使用 -O 运行也必须检查）
def factorial_safe(n):
    if not isinstance(n, int) or n < 0:
        raise ValueError("n 必须是非负整数")    # 生产环境也必须检查
    if n <= 1:
        return 1
    return n * factorial_safe(n - 1)

# EAFP：先尝试，失败时处理（推荐）
def get_config_value(config, key, default=None):
    try:
        return config[key]              # 直接访问，失败再处理
    except KeyError:
        return default

# LBYL：先检查，再访问
def get_config_value_lbyl(config, key, default=None):
    if key in config:                   # 先检查
        return config[key]              # 再访问
    else:
        return default

# EAFP 避免竞态条件的经典例子
import os
# LBYL 有竞态条件：检查和创建之间文件可能已被其他进程创建
if not os.path.exists("file.txt"):     # 检查时不存在
    with open("file.txt", "w") as f:   # 但创建时可能已存在
        f.write("data")

# EAFP 无竞态条件
try:
    with open("file.txt", "x") as f:   # "x" 模式：排他创建，原子操作
        f.write("data")
except FileExistsError:
    print("文件已存在")

import unittest

class TestExceptionHandling(unittest.TestCase):
    def test_divide_by_zero(self):
        with self.assertRaises(ZeroDivisionError):
            divide(10, 0)

    def test_age_validation(self):
        with self.assertRaises(AgeError):
            check_age(150)

# 使用 pytest 测试异常
def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

# 陷阱：bare except 会捕获 KeyboardInterrupt 和 SystemExit
# 导致你无法用 Ctrl+C 终止程序！
while True:
    try:
        do_work()
    except:                  # 危险！连 Ctrl+C 都会被捕获
        pass                 # 程序无法正常退出

# 正确做法：至少使用 except Exception
while True:
    try:
        do_work()
    except Exception as e:   # 不会捕获 KeyboardInterrupt 和 SystemExit
        logging.error(f"工作失败: {e}")

# 陷阱：finally 中的 return 会吞掉 try 块中的异常！
def dangerous_function():
    try:
        raise ValueError("重要错误！")
    finally:
        return 42            # 这个 return 会覆盖异常，导致异常丢失！

result = dangerous_function()
print(result)                # 42 —— 异常被吞掉了！没有任何报错！


# 同样，finally 中的 return 也会覆盖 try 中的 return
def another_dangerous_function():
    try:
        return 100
    finally:
        return 0             # 这个 return 覆盖了 try 中的 return 100

result = another_dangerous_function()
print(result)                # 0 —— 不是 100！


# 正确做法：不要在 finally 中使用 return
def safe_function():
    result = None
    try:
        result = do_something()
    except ValueError as e:
        logging.error(f"错误: {e}")
        raise                # 让异常正常传播
    finally:
        cleanup()            # finally 只做清理，不做 return
    return result

# 陷阱：生成器中的异常可能在不经意间被传播或吞掉
def bad_generator():
    yield 1
    raise ValueError("生成器中的错误")  # 异常会传播给调用者
    yield 2                              # 永远不会执行

gen = bad_generator()
print(next(gen))            # 1
try:
    next(gen)               # 触发 ValueError
except ValueError as e:
    print(f"捕获: {e}")     # 捕获: 生成器中的错误


# 陷阱：生成器的 close() 方法会抛出 GeneratorExit
def my_generator():
    try:
        while True:
            yield 1
    except GeneratorExit:           # GeneratorExit 是 BaseException 的子类
        print("生成器被关闭")        # 不是 Exception 的子类！
        return                      # 清理并退出

gen = my_generator()
next(gen)                   # 1
gen.close()                  # 触发 GeneratorExit → 输出 "生成器被关闭"


# 陷阱：在生成器中用 except Exception 不会捕获 GeneratorExit
def another_generator():
    try:
        while True:
            yield 1
    except Exception:               # 不会捕获 GeneratorExit！
        print("这行不会执行")
    # GeneratorExit 会穿透 except Exception，生成器被关闭

gen = another_generator()
next(gen)
gen.close()                  # GeneratorExit 穿透 except Exception

# 反模式：用异常代替正常的流程控制
def find_item(lst, target):
    """用异常控制流程——不推荐"""
    try:
        index = lst.index(target)    # index() 在找不到时抛出 ValueError
        return index
    except ValueError:
        return -1                    # 用异常来表达"未找到"这个正常结果

# 更好的做法：使用返回值表达正常逻辑
def find_item_better(lst, target):
    """用条件判断——推荐用于这种正常逻辑"""
    if target in lst:
        return lst.index(target)
    return -1


# 另一个反模式：用异常实现循环退出
def read_until_empty(lst):
    """用异常控制循环——不推荐"""
    i = 0
    while True:
        try:
            yield lst[i]
            i += 1
        except IndexError:
            break                   # 用 IndexError 表示"遍历结束"

# 正确做法：使用 for 循环
def read_until_empty_better(lst):
    """正常迭代——推荐"""
    for item in lst:
        yield item

# 陷阱：except Exception 不会捕获 KeyboardInterrupt
import time

def long_running_task():
    try:
        while True:
            time.sleep(1)
            print("工作中...")
    except Exception:               # Ctrl+C 不会被这里捕获！
        print("异常被捕获")          # 这行不会执行

# 运行后按 Ctrl+C，程序直接崩溃而不是优雅退出
# KeyboardInterrupt 继承自 BaseException，不是 Exception

# 正确做法：如需捕获中断信号，显式捕获 KeyboardInterrupt
def long_running_task_safe():
    try:
        while True:
            time.sleep(1)
            print("工作中...")
    except KeyboardInterrupt:       # 显式捕获中断信号
        print("\n收到中断信号，正在优雅退出...")
        # 执行清理工作
    finally:
        print("资源已释放")

if key in my_dict:
    value = my_dict[key]
else:
    value = default_value

try:
    value = my_dict[key]
except KeyError:
    value = default_value