import time
import functools
from typing import Callable, Type, Tuple
def retry_with_backoff(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    exponential_base: float = 2.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
    """
    使用指數(shù)退避算法重試函數(shù)的裝飾器。
    參數(shù):
        max_attempts: 最大重試次數(shù)
        base_delay: 初始等待時間（秒）
        exponential_base: 指數(shù)基數(shù)（2.0 表示每次等待時間翻倍）
        exceptions: 需要捕獲并重試的異常類型元組
    """
    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            # 嘗試執(zhí)行函數(shù)，最多執(zhí)行 max_attempts 次
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    # 如果還有剩余重試次數(shù)，則計算等待時間并休眠
                    if attempt < max_attempts - 1:
                        delay = base_delay * (exponential_base ** attempt)
                        print(f"第 {attempt + 1} 次嘗試失敗: {e}")
                        print(f"將在 {delay:.1f} 秒后重試...")
                        time.sleep(delay)
                    else:
                        # 所有嘗試均失敗，拋出最后一次異常
                        print(f"所有 {max_attempts} 次嘗試均失敗")
                        raise last_exception
            # 理論上不會執(zhí)行到這里，因為要么返回結(jié)果，要么拋出異常
            # 但為了類型檢查器的完整性，保留此結(jié)構(gòu)
        return wrapper
    return decorator

if __name__ == "__main__":
    counter = 0
    @retry_with_backoff(max_attempts=3, base_delay=0.5, exceptions=(ValueError,))
    def unstable_function():
        global counter
        counter += 1
        print(f"正在執(zhí)行函數(shù)，當前調(diào)用次數(shù): {counter}")
        if counter < 3:
            raise ValueError("模擬臨時性錯誤")
        return "成功！"
    try:
        result = unstable_function()
        print(f"最終結(jié)果: {result}")
    except Exception as e:
        print(f"未處理的異常: {e}")

from typing import Any, Callable, Dict, List, Optional
class ValidationError(Exception):
    """當驗證失敗時拋出的異常。"""
    def __init__(self, field: str, errors: List[str]):
        self.field = field
        self.errors = errors
        # 調(diào)用父類構(gòu)造函數(shù)，設置異常消息為字段名和錯誤列表的組合
        super().__init__(f"{field}: {', '.join(errors)}")

def validate_input(
    value: Any,
    field_name: str,
    rules: Dict[str, Callable[[Any], bool]],
    messages: Optional[Dict[str, str]] = None
) -> Any:
    """
    根據(jù)多條規(guī)則驗證輸入。
    參數(shù):
        value: 待驗證的值
        field_name: 字段名稱，用于錯誤提示
        rules: 字典，鍵為規(guī)則名稱，值為驗證函數(shù) (接受值，返回布爾值)
        messages: 可選字典，鍵為規(guī)則名稱，值為自定義錯誤消息
    返回:
        如果驗證通過，返回原始值
    異常:
        ValidationError: 如果任何規(guī)則驗證失敗
    """
    if messages is None:
        messages = {}
    errors = []
    # 遍歷所有驗證規(guī)則
    for rule_name, rule_func in rules.items():
        try:
            # 執(zhí)行驗證函數(shù)
            if not rule_func(value):
                # 如果驗證失敗，獲取自定義錯誤消息或默認消息
                error_msg = messages.get(
                    rule_name, f"未通過驗證規(guī)則: {rule_name}"
                )
                errors.append(error_msg)
        except Exception as e:
            # 如果在執(zhí)行驗證規(guī)則過程中發(fā)生異常（例如類型轉(zhuǎn)換錯誤），也視為驗證失敗
            errors.append(f"規(guī)則 '{rule_name}' 執(zhí)行出錯: {str(e)}")
    # 如果存在錯誤，拋出驗證異常
    if errors:
        raise ValidationError(field_name, errors)
    return value

# --- 可復用的驗證規(guī)則工廠函數(shù) ---
def not_empty(value: Any) -> bool:
    """檢查值是否非空（針對字符串去除空白后）"""
    if isinstance(value, str):
        return bool(value.strip())
    return bool(value)
def min_length(min_len: int) -> Callable[[Any], bool]:
    """生成一個檢查最小長度的驗證函數(shù)"""
    def _validator(value: Any) -> bool:
        try:
            return len(str(value)) >= min_len
        except Exception:
            return False
    return _validator
def max_length(max_len: int) -> Callable[[Any], bool]:
    """生成一個檢查最大長度的驗證函數(shù)"""
    def _validator(value: Any) -> bool:
        try:
            return len(str(value)) <= max_len
        except Exception:
            return False
    return _validator
def in_range(min_val: float, max_val: float) -> Callable[[Any], bool]:
    """生成一個檢查數(shù)值范圍的驗證函數(shù)"""
    def _validator(value: Any) -> bool:
        try:
            num_value = float(value)
            return min_val <= num_value <= max_val
        except (ValueError, TypeError):
            # 如果無法轉(zhuǎn)換為浮點數(shù)，則驗證失敗
            return False
    return _validator

try:
    username = validate_input(
        "ab",  # 輸入值
        "username",  # 字段名
        {
            "not_empty": not_empty,
            "min_length": min_length(3),
            "max_length": max_length(20),
        },
        messages={
            "not_empty": "用戶名不能為空",
            "min_length": "用戶名長度至少為 3 個字符",
            "max_length": "用戶名長度不能超過 20 個字符",
        }
    )
    print(f"驗證通過的用戶名: {username}")
except ValidationError as e:
    print(f"驗證失敗: {e}")

驗證失敗: username: 用戶名長度至少為 3 個字符

from typing import Any, Optional, List, Union
def safe_get(
    data: dict,
    path: Union[str, List[str]],
    default: Any = None,
    separator: str = "."
) -> Any:
    """
    從嵌套字典中安全地獲取值。
    參數(shù):
        data: 要訪問的字典
        path: 點分路徑 (例如 "user.address.city") 或鍵名列表
        default: 如果路徑不存在時返回的默認值
        separator: 用于分割路徑字符串的分隔符 (默認: ".")
    返回:
        路徑對應的值，如果未找到則返回 default
    """
    # 如果路徑是字符串，將其分割為鍵名列表
    if isinstance(path, str):
        keys = path.split(separator)
    else:
        keys = path
    current = data
    for key in keys:
        try:
            # 處理列表索引：如果當前層級是列表，嘗試將鍵轉(zhuǎn)換為整數(shù)索引
            if isinstance(current, list):
                try:
                    index = int(key)
                    current = current[index]
                except (ValueError, TypeError, IndexError):
                    # 如果轉(zhuǎn)換失敗或索引越界，返回默認值
                    return default
            elif isinstance(current, dict):
                # 如果當前層級是字典，正常按鍵獲取
                current = current[key]
            else:
                # 如果當前層級既不是字典也不是列表（例如是字符串或數(shù)字），無法繼續(xù)深入
                return default
        except (KeyError, IndexError, TypeError):
            # 捕獲鍵錯誤、索引錯誤或類型錯誤，返回默認值
            return default
    return current

def safe_set(
    data: dict,
    path: Union[str, List[str]],
    value: Any,
    separator: str = ".",
    create_missing: bool = True
) -> bool:
    """
    安全地在嵌套字典中設置值。
    參數(shù):
        data: 要修改的字典
        path: 點分路徑或鍵名列表
        value: 要設置的值
        separator: 用于分割路徑字符串的分隔符
        create_missing: 是否創(chuàng)建缺失的中間字典節(jié)點
    返回:
        如果設置成功返回 True，否則返回 False
    """
    # 將路徑字符串分割為鍵名列表
    if isinstance(path, str):
        keys = path.split(separator)
    else:
        keys = path
    # 如果路徑為空，無法設置
    if not keys:
        return False
    current = data
    # 遍歷到最后一個鍵的父節(jié)點
    for key in keys[:-1]:
        # 如果當前節(jié)點不存在
        if key not in current:
            if create_missing:
                # 如果允許創(chuàng)建缺失節(jié)點，則創(chuàng)建一個新的空字典
                current[key] = {}
            else:
                # 如果不允許創(chuàng)建，則設置失敗
                return False
        # 移動到下一層
        next_node = current[key]
        # 檢查下一層是否是字典，因為只有字典才能包含子鍵
        # 注意：這里不支持在列表中自動插入元素，只支持字典嵌套
        if not isinstance(next_node, dict):
            # 如果中間節(jié)點存在但不是字典（例如是列表或基本類型），無法繼續(xù)深入設置
            return False
        current = next_node
    # 設置最終的值
    try:
        final_key = keys[-1]
        # 如果最后一層是列表，嘗試作為索引設置（可選功能，這里簡化為僅支持字典鍵）
        # 為了保持簡單和一致性，safe_set 主要針對字典結(jié)構(gòu)。
        # 如果用戶希望設置列表元素，通常需要先確保列表存在且長度足夠，這比較復雜。
        # 此處我們假設最終目標是一個字典的鍵。
        current[final_key] = value
        return True
    except Exception:
        return False

if __name__ == "__main__":
    # 示例嵌套數(shù)據(jù)
    user_data = {
        "user": {
            "name": "Anna",
            "address": {
                "city": "San Francisco",
                "zip": "94105"
            },
            "orders": [
                {"id": 1, "total": 99.99},
                {"id": 2, "total": 149.50}
            ]
        }
    }
    print("--- Safe Get 測試 ---")
    # 1. 獲取存在的城市
    city = safe_get(user_data, "user.address.city")
    print(f"城市: {city}")
    # 2. 獲取不存在的國家，使用默認值
    country = safe_get(user_data, "user.address.country", default="Unknown")
    print(f"國家: {country}")
    # 3. 獲取列表中的第一個訂單總額
    # 注意：路徑中的 "0" 會被識別為列表索引
    first_order_total = safe_get(user_data, "user.orders.0.total")
    print(f"第一筆訂單總額: ${first_order_total}")
    # 4. 獲取深層不存在的路徑
    missing = safe_get(user_data, "user.profile.bio", default="No bio")
    print(f"個人簡介: {missing}")
    # 5. 嘗試在非字典節(jié)點上繼續(xù)獲取 (orders 是列表，不能直接用 .id 獲取所有 ID，只能索引)
    invalid_path = safe_get(user_data, "user.orders.id", default="Invalid Path")
    print(f"無效路徑結(jié)果: {invalid_path}")
    print("\n--- Safe Set 測試 ---")
    # 1. 在空字典中創(chuàng)建深層結(jié)構(gòu)
    new_data = {}
    success = safe_set(new_data, "user.settings.theme", "dark")
    print(f"設置主題成功: {success}")
    print(f"新數(shù)據(jù)結(jié)構(gòu): {new_data}")
    # 2. 修改已存在的值
    success = safe_set(user_data, "user.address.city", "New York")
    print(f"修改城市成功: {success}")
    print(f"新城市: {safe_get(user_data, 'user.address.city')}")
    # 3. 嘗試在不允許創(chuàng)建缺失節(jié)點的情況下設置
    temp_data = {"a": 1}
    success = safe_set(temp_data, "a.b.c", "value", create_missing=False)
    print(f"在不允許創(chuàng)建缺失節(jié)點時設置深層路徑成功: {success}") # 應為 False，因為 a 是整數(shù)不是字典
    # 4. 覆蓋非字典節(jié)點
    overwrite_data = {"x": {"y": "old"}}
    # 先將 y 設置為一個字典，以便后續(xù)可以在其下設置 z
    safe_set(overwrite_data, "x.y", {"z": "new"})
    success = safe_set(overwrite_data, "x.y.z", "final_value")
    print(f"覆蓋后設置深層值成功: {success}")
    print(f"最終值: {safe_get(overwrite_data, 'x.y.z')}")

--- Safe Get 測試 ---
城市: San Francisco
國家: Unknown
第一筆訂單總額: $99.99
個人簡介: No bio
無效路徑結(jié)果: Invalid Path
--- Safe Set 測試 ---
設置主題成功: True
新數(shù)據(jù)結(jié)構(gòu): {'user': {'settings': {'theme': 'dark'}}}
修改城市成功: True
新城市: New York
在不允許創(chuàng)建缺失節(jié)點時設置深層路徑成功: False
覆蓋后設置深層值成功: True
最終值: final_value

import threading
import functools
import time
from typing import Callable, Optional, Any
class TimeoutError(Exception):
    """當操作超過指定超時時限后拋出的異常。"""
    pass
def timeout(seconds: float, error_message: Optional[str] = None):
    """
    一個用于強制函數(shù)執(zhí)行超時的裝飾器。
    注意：此實現(xiàn)使用線程。由于 Python 的 GIL 和線程機制，
    它無法真正“殺死”正在運行的線程，只能停止等待并拋出異常。
    后臺線程可能會繼續(xù)運行直到完成，但其結(jié)果將被忽略。
    參數(shù):
        seconds: 最大執(zhí)行時間（秒）
        error_message: 超時時的自定義錯誤消息
    返回:
        裝飾器函數(shù)
    """
    if seconds < 0:
        raise ValueError("超時時間不能為負數(shù)")
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # 使用列表來存儲結(jié)果或異常，因為列表是可變對象，可以在閉包中修改
            # 初始化為 None，表示尚未完成
            result_container = [None]
            exception_container = [None]
            def target():
                try:
                    # 執(zhí)行目標函數(shù)
                    result_container[0] = func(*args, **kwargs)
                except Exception as e:
                    # 捕獲函數(shù)內(nèi)部拋出的任何異常
                    exception_container[0] = e
            # 創(chuàng)建并啟動線程
            thread = threading.Thread(target=target)
            thread.daemon = True  # 設置為守護線程，主程序退出時自動結(jié)束
            thread.start()
            # 等待線程完成，最多等待 seconds 秒
            thread.join(timeout=seconds)
            # 檢查線程是否仍在運行（即是否超時）
            if thread.is_alive():
                # 如果超時，拋出 TimeoutError
                # 注意：線程實際上仍在后臺運行，但我們不再關(guān)心其結(jié)果
                raise TimeoutError(
                    error_message or f"操作在 {seconds} 秒后超時"
                )
            # 如果線程已結(jié)束，檢查是否有異常
            if exception_container[0] is not None:
                raise exception_container[0]
            # 返回正常結(jié)果
            return result_container[0]
        return wrapper
    return decorator

if __name__ == "__main__":
    print("--- 測試 1: 模擬慢速數(shù)據(jù)庫查詢 (應超時) ---")
    @timeout(2, error_message="查詢耗時過長")
    def slow_database_query():
        """模擬一個慢速查詢。"""
        time.sleep(5)
        return "查詢結(jié)果"
    try:
        result = slow_database_query()
        print(f"結(jié)果: {result}")
    except TimeoutError as e:
        print(f"超時錯誤: {e}")
    print("\n--- 測試 2: 模擬快速數(shù)據(jù)獲取 (應成功) ---")
    @timeout(3)
    def fetch_data():
        """模擬一個快速操作。"""
        time.sleep(1)
        return {"data": "value"}
    try:
        data = fetch_data()
        print(f"成功: {data}")
    except TimeoutError as e:
        print(f"超時錯誤: {e}")
    print("\n--- 測試 3: 函數(shù)內(nèi)部拋出異常 ---")
    @timeout(5)
    def faulty_function():
        """模擬一個會拋出異常的函數(shù)。"""
        raise ValueError("內(nèi)部錯誤")
    try:
        faulty_function()
    except ValueError as e:
        print(f"捕獲到內(nèi)部異常: {e}")
    except TimeoutError as e:
        print(f"超時錯誤: {e}")
    print("\n--- 測試 4: 剛好在超時前完成 ---")
    @timeout(2)
    def just_in_time():
        """模擬一個剛好在超時前完成的操作。"""
        time.sleep(1.9)
        return "及時完成"
    try:
        result = just_in_time()
        print(f"結(jié)果: {result}")
    except TimeoutError as e:
        print(f"超時錯誤: {e}")

--- 測試 1: 模擬慢速數(shù)據(jù)庫查詢 (應超時) ---
超時錯誤: 查詢耗時過長
--- 測試 2: 模擬快速數(shù)據(jù)獲取 (應成功) ---
成功: {'data': 'value'}
--- 測試 3: 函數(shù)內(nèi)部拋出異常 ---
捕獲到內(nèi)部異常: 內(nèi)部錯誤
--- 測試 4: 剛好在超時前完成 ---
結(jié)果: 及時完成

from contextlib import contextmanager
from typing import Callable, Any, Optional
import traceback
@contextmanager
def managed_resource(
    acquire: Callable[[], Any],
    release: Callable[[Any], None],
    on_error: Optional[Callable[[Exception, Any], None]] = None,
    suppress_errors: bool = False
):
    """
    用于自動獲取和清理資源的上下文管理器。
    參數(shù):
        acquire: 獲取資源的函數(shù)
        release: 釋放資源的函數(shù)
        on_error: 可選的錯誤處理函數(shù)，接收異常對象和資源對象
        suppress_errors: 是否在清理后抑制異常（即不重新拋出）
    注意:
        如果發(fā)生異常，執(zhí)行順序為:
. 調(diào)用 on_error (如果提供且資源已獲取)
. 如果 suppress_errors 為 False，則重新拋出原始異常
. 始終執(zhí)行 release (在 finally 塊中)
    """
    resource = None
    try:
        # 獲取資源
        resource = acquire()
        # 將資源 yield 給 with 塊使用
        yield resource
    except Exception as e:
        # 如果在 with 塊中發(fā)生異常
        if on_error and resource is not None:
            try:
                # 調(diào)用自定義錯誤處理器
                on_error(e, resource)
            except Exception as handler_error:
                # 如果錯誤處理器本身出錯，打印警告，但不掩蓋原始異常
                print(f"錯誤處理器執(zhí)行出錯: {handler_error}")
                traceback.print_exc()
        # 根據(jù)配置決定是否抑制異常
        if not suppress_errors:
            raise
    finally:
        # 無論是否發(fā)生異常，都嘗試釋放資源
        if resource is not None:
            try:
                release(resource)
            except Exception as cleanup_error:
                # 如果清理過程中出錯，打印錯誤信息
                print(f"資源清理過程中出錯: {cleanup_error}")
                traceback.print_exc()

# --- 輔助類：資源追蹤器 ---
class ResourceTracker:
    """用于追蹤資源操作的輔助類。"""
    def __init__(self, name: str, verbose: bool = True):
        self.name = name
        self.verbose = verbose
        self.operations = []
    def log(self, operation: str):
        """記錄操作并可選地打印。"""
        self.operations.append(operation)
        if self.verbose:
            print(f"[{self.name}] {operation}")
    def acquire(self):
        """模擬獲取資源。"""
        self.log("正在獲取資源")
        return self
    def release(self):
        """模擬釋放資源。"""
        self.log("正在釋放資源")
    def use(self, action: str):
        """模擬使用資源執(zhí)行操作。"""
        self.log(f"使用資源執(zhí)行: {action}")

# --- 測試示例 ---
if __name__ == "__main__":
    print("--- 測試 1: 帶錯誤處理的正常異常流程 ---")
    tracker1 = ResourceTracker("Database")
    def error_handler(exception, resource):
        resource.log(f"捕獲到錯誤: {exception}")
        resource.log("嘗試執(zhí)行回滾操作")
    try:
        with managed_resource(
            acquire=lambda: tracker1.acquire(),
            release=lambda r: r.release(),
            on_error=error_handler
        ) as db:
            db.use("INSERT INTO users")
            raise ValueError("重復條目錯誤")
    except ValueError as e:
        print(f"主程序捕獲到異常: {e}")
    print("\n--- 測試 2: 成功執(zhí)行無異常 ---")
    tracker2 = ResourceTracker("FileHandle")
    try:
        with managed_resource(
            acquire=lambda: tracker2.acquire(),
            release=lambda r: r.release()
        ) as file:
            file.use("Write data")
        print("操作成功完成")
    except Exception as e:
        print(f"意外異常: {e}")
    print("\n--- 測試 3: 抑制異常 ---")
    tracker3 = ResourceTracker("NetworkSocket")
    try:
        with managed_resource(
            acquire=lambda: tracker3.acquire(),
            release=lambda r: r.release(),
            suppress_errors=True
        ) as sock:
            sock.use("Send packet")
            raise ConnectionError("連接丟失")
    except Exception as e:
        print(f"此異常應被抑制，不應打印: {e}")
    else:
        print("異常已被抑制，程序繼續(xù)執(zhí)行")
    print("\n--- 測試 4: 清理過程中出錯 ---")
    tracker4 = ResourceTracker("BadResource")
    def bad_release(r):
        r.log("嘗試釋放... 失敗!")
        raise RuntimeError("釋放失敗")
    try:
        with managed_resource(
            acquire=lambda: tracker4.acquire(),
            release=bad_release
        ) as res:
            res.use("Do something")
    except Exception as e:
        print(f"主程序未捕獲到清理異常（因為它在 finally 中被捕獲并打?。? {type(e).__name__}")

--- 測試 1: 帶錯誤處理的正常異常流程 ---
[Database] 正在獲取資源
[Database] 使用資源執(zhí)行: INSERT INTO users
[Database] 捕獲到錯誤: 重復條目錯誤
[Database] 嘗試執(zhí)行回滾操作
[Database] 正在釋放資源
主程序捕獲到異常: 重復條目錯誤
--- 測試 2: 成功執(zhí)行無異常 ---
[FileHandle] 正在獲取資源
[FileHandle] 使用資源執(zhí)行: Write data
[FileHandle] 正在釋放資源
操作成功完成
--- 測試 3: 抑制異常 ---
[NetworkSocket] 正在獲取資源
[NetworkSocket] 使用資源執(zhí)行: Send packet
[NetworkSocket] 正在釋放資源
異常已被抑制，程序繼續(xù)執(zhí)行
--- 測試 4: 清理過程中出錯 ---
[BadResource] 正在獲取資源
[BadResource] 使用資源執(zhí)行: Do something
[BadResource] 嘗試釋放... 失敗!
資源清理過程中出錯: 釋放失敗
Traceback (most recent call last):
  File "...", line ..., in managed_resource
    release(resource)
  File "...", line ..., in bad_release
    raise RuntimeError("釋放失敗")
RuntimeError: 釋放失敗
主程序未捕獲到清理異常（因為它在 finally 中被捕獲并打?。? Exception