with expression as variable:
    # 代碼塊
    pass

with expression1 as var1, expression2 as var2:
    # 代碼塊
    pass

# 傳統(tǒng)方式：手動(dòng)打開和關(guān)閉文件
file = open('example.txt', 'r')
content = file.read()
print(content)
file.close()  # 必須手動(dòng)關(guān)閉文件

def read_file_traditional(filename):
    """傳統(tǒng)方式讀取文件，展示潛在問題"""
    file = open(filename, 'r')
    try:
        content = file.read()
        # 模擬一個(gè)可能導(dǎo)致異常的操作
        result = 1 / 0  # 這會(huì)引發(fā)ZeroDivisionError
        return content
    finally:
        file.close()  # 即使有異常也會(huì)執(zhí)行

# 調(diào)用函數(shù)
try:
    read_file_traditional('nonexistent.txt')
except Exception as e:
    print(f"捕獲到異常: {e}")

# 使用with語(yǔ)句讀取文件
with open('example.txt', 'r') as file:
    content = file.read()
    print(content)
# 文件在此處自動(dòng)關(guān)閉，無需手動(dòng)調(diào)用close()

# 寫入文件
with open('output.txt', 'w') as file:
    file.write('Hello, World!')
# 文件自動(dòng)關(guān)閉

class MyContextManager:
    def __enter__(self):
        print("進(jìn)入上下文")
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        print("退出上下文")
        if exc_type is not None:
            print(f"發(fā)生了異常: {exc_type.__name__}: {exc_value}")
        return False  # 返回False表示不抑制異常

# 使用自定義上下文管理器
with MyContextManager() as cm:
    print("在上下文中執(zhí)行代碼")
    # raise ValueError("測(cè)試異常")  # 取消注釋測(cè)試異常處理

class ExceptionSuppressor:
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is not None:
            print(f"捕獲并抑制了異常: {exc_type.__name__}: {exc_value}")
            return True  # 抑制異常
        return False

# 測(cè)試異常抑制
with ExceptionSuppressor():
    print("開始執(zhí)行...")
    raise ValueError("這是一個(gè)測(cè)試異常")
    print("這行不會(huì)執(zhí)行")

print("程序繼續(xù)執(zhí)行")

# 讀取文本文件
def read_text_file(filename):
    with open(filename, 'r', encoding='utf-8') as file:
        return file.read()

# 寫入JSON數(shù)據(jù)
import json

def write_json_data(data, filename):
    with open(filename, 'w', encoding='utf-8') as file:
        json.dump(data, file, indent=2, ensure_ascii=False)

# 讀取JSON數(shù)據(jù)
def read_json_data(filename):
    with open(filename, 'r', encoding='utf-8') as file:
        return json.load(file)

# 示例使用
data = {
    "name": "張三",
    "age": 30,
    "city": "北京"
}

write_json_data(data, 'user.json')
loaded_data = read_json_data('user.json')
print(loaded_data)

# 復(fù)制二進(jìn)制文件
def copy_binary_file(source, destination):
    with open(source, 'rb') as src_file:
        with open(destination, 'wb') as dst_file:
            while True:
                chunk = src_file.read(1024)  # 每次讀取1KB
                if not chunk:
                    break
                dst_file.write(chunk)

# 使用示例
# copy_binary_file('source.jpg', 'destination.jpg')

# 同時(shí)打開多個(gè)文件進(jìn)行處理
def process_multiple_files(input_file, output_file, log_file):
    with open(input_file, 'r') as infile, \
         open(output_file, 'w') as outfile, \
         open(log_file, 'w') as logfile:
        
        for line_num, line in enumerate(infile, 1):
            try:
                # 處理每一行
                processed_line = line.strip().upper()
                outfile.write(processed_line + '\n')
                
                if line_num % 100 == 0:
                    logfile.write(f"已處理 {line_num} 行\(zhòng)n")
                    
            except Exception as e:
                logfile.write(f"第 {line_num} 行處理出錯(cuò): {e}\n")

# 創(chuàng)建測(cè)試數(shù)據(jù)
test_data = [f"這是第{i}行數(shù)據(jù)\n" for i in range(1, 1001)]

with open('input.txt', 'w') as f:
    f.writelines(test_data)

# 處理文件
process_multiple_files('input.txt', 'output.txt', 'process.log')

import os
import time

class FileLock:
    def __init__(self, lock_file):
        self.lock_file = lock_file
    
    def __enter__(self):
        # 等待獲取鎖
        while os.path.exists(self.lock_file):
            time.sleep(0.1)
        
        # 創(chuàng)建鎖文件
        with open(self.lock_file, 'w') as f:
            f.write(str(os.getpid()))
        
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        # 刪除鎖文件
        if os.path.exists(self.lock_file):
            os.remove(self.lock_file)

# 使用文件鎖
def critical_operation():
    with FileLock('operation.lock'):
        print("執(zhí)行關(guān)鍵操作...")
        time.sleep(2)  # 模擬耗時(shí)操作
        print("操作完成")

# 在不同線程或進(jìn)程中調(diào)用critical_operation()

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print(f"連接到數(shù)據(jù)庫(kù): {self.connection_string}")
        # 這里應(yīng)該是實(shí)際的數(shù)據(jù)庫(kù)連接代碼
        self.connection = f"連接對(duì)象({self.connection_string})"
        return self.connection
    
    def __exit__(self, exc_type, exc_value, traceback):
        print("關(guān)閉數(shù)據(jù)庫(kù)連接")
        # 這里應(yīng)該是實(shí)際的關(guān)閉連接代碼
        self.connection = None

# 使用數(shù)據(jù)庫(kù)連接
def query_database():
    with DatabaseConnection("postgresql://localhost/mydb") as conn:
        print(f"使用連接: {conn}")
        # 執(zhí)行數(shù)據(jù)庫(kù)查詢...
        return "查詢結(jié)果"

result = query_database()

import urllib.request
from contextlib import contextmanager

@contextmanager
def web_request(url):
    print(f"發(fā)起請(qǐng)求: {url}")
    response = None
    try:
        response = urllib.request.urlopen(url)
        yield response
    except Exception as e:
        print(f"請(qǐng)求失敗: {e}")
        raise
    finally:
        if response:
            response.close()
            print("響應(yīng)已關(guān)閉")

# 使用網(wǎng)絡(luò)請(qǐng)求上下文管理器
def fetch_web_content(url):
    try:
        with web_request(url) as response:
            content = response.read().decode('utf-8')
            return content[:200] + "..."  # 只返回前200個(gè)字符
    except Exception as e:
        return f"獲取內(nèi)容失敗: {e}"

# 注意：實(shí)際使用時(shí)請(qǐng)?zhí)鎿Q為可訪問的URL
# content = fetch_web_content('https://httpbin.org/get')
# print(content)

from contextlib import contextmanager
import time

@contextmanager
def timer():
    start_time = time.time()
    print("計(jì)時(shí)開始")
    try:
        yield
    finally:
        end_time = time.time()
        print(f"耗時(shí): {end_time - start_time:.2f} 秒")

# 使用timer上下文管理器
with timer():
    time.sleep(1)  # 模擬耗時(shí)操作
    print("執(zhí)行一些任務(wù)...")

@contextmanager
def temporary_change_dir(new_dir):
    import os
    old_dir = os.getcwd()
    try:
        os.chdir(new_dir)
        yield
    finally:
        os.chdir(old_dir)

# 使用臨時(shí)目錄切換
# with temporary_change_dir('/tmp'):
#     print(f"當(dāng)前目錄: {os.getcwd()}")
# print(f"恢復(fù)目錄: {os.getcwd()}")

from contextlib import suppress
import os

# 抑制特定異常
with suppress(FileNotFoundError):
    with open('nonexistent.txt', 'r') as f:
        content = f.read()
    print("文件內(nèi)容:", content)

# 抑制多個(gè)異常類型
with suppress(FileNotFoundError, PermissionError):
    os.remove('protected_file.txt')
    print("文件刪除成功")

from contextlib import redirect_stdout, redirect_stderr
import io
import sys

# 重定向標(biāo)準(zhǔn)輸出
output_buffer = io.StringIO()
with redirect_stdout(output_buffer):
    print("這條消息被重定向了")
    print("這條也是")

captured_output = output_buffer.getvalue()
print(f"捕獲的輸出: {captured_output}")

# 重定向標(biāo)準(zhǔn)錯(cuò)誤
error_buffer = io.StringIO()
with redirect_stderr(error_buffer):
    print("錯(cuò)誤信息", file=sys.stderr)

captured_error = error_buffer.getvalue()
print(f"捕獲的錯(cuò)誤: {captured_error}")

import time
import tempfile
import os

def traditional_file_handling(filename, iterations=1000):
    """傳統(tǒng)文件處理方式"""
    start_time = time.time()
    
    for i in range(iterations):
        file = open(filename, 'a')
        file.write(f"Line {i}\n")
        file.close()
    
    return time.time() - start_time

def with_statement_handling(filename, iterations=1000):
    """使用with語(yǔ)句的文件處理方式"""
    start_time = time.time()
    
    for i in range(iterations):
        with open(filename, 'a') as file:
            file.write(f"Line {i}\n")
    
    return time.time() - start_time

def contextlib_handling(filename, iterations=1000):
    """使用contextlib的文件處理方式"""
    from contextlib import closing
    
    start_time = time.time()
    
    for i in range(iterations):
        with closing(open(filename, 'a')) as file:
            file.write(f"Line {i}\n")
    
    return time.time() - start_time

# 創(chuàng)建臨時(shí)文件進(jìn)行測(cè)試
temp_file = tempfile.NamedTemporaryFile(delete=False)
temp_filename = temp_file.name
temp_file.close()

try:
    # 執(zhí)行性能測(cè)試
    traditional_time = traditional_file_handling(temp_filename)
    with_time = with_statement_handling(temp_filename)
    contextlib_time = contextlib_handling(temp_filename)
    
    print("性能對(duì)比結(jié)果:")
    print(f"傳統(tǒng)方式: {traditional_time:.4f} 秒")
    print(f"with語(yǔ)句: {with_time:.4f} 秒")
    print(f"contextlib: {contextlib_time:.4f} 秒")
    
finally:
    # 清理臨時(shí)文件
    os.unlink(temp_filename)

# 推薦：明確指定編碼
with open('file.txt', 'r', encoding='utf-8') as f:
    content = f.read()

# 不推薦：依賴系統(tǒng)默認(rèn)編碼
with open('file.txt', 'r') as f:
    content = f.read()

# 推薦：使用原始字符串處理路徑
with open(r'C:\path\to\file.txt', 'r') as f:
    content = f.read()

# 處理大文件時(shí)使用迭代器
def process_large_file(filename):
    with open(filename, 'r', encoding='utf-8') as f:
        for line in f:  # 逐行讀取，避免內(nèi)存溢出
            process_line(line.strip())

def process_line(line):
    # 處理單行數(shù)據(jù)
    pass

def robust_file_operation(filename):
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        print(f"文件 {filename} 不存在")
        return None
    except PermissionError:
        print(f"沒有權(quán)限訪問文件 {filename}")
        return None
    except UnicodeDecodeError:
        print(f"文件 {filename} 編碼錯(cuò)誤，嘗試其他編碼")
        try:
            with open(filename, 'r', encoding='gbk') as f:
                return f.read()
        except Exception:
            print("無法解碼文件")
            return None
    except Exception as e:
        print(f"讀取文件時(shí)發(fā)生未知錯(cuò)誤: {e}")
        return None

import gc
import weakref

class ResourceTracker:
    def __init__(self):
        self.resources = weakref.WeakSet()
    
    def track(self, resource):
        self.resources.add(resource)
        return resource
    
    def get_leaked_resources(self):
        # 強(qiáng)制垃圾回收
        gc.collect()
        return len(self.resources)

# 全局資源跟蹤器
tracker = ResourceTracker()

class TrackedFile:
    def __init__(self, filename, mode):
        self.file = open(filename, mode)
        self.filename = filename
        print(f"打開文件: {filename}")
    
    def __enter__(self):
        return tracker.track(self.file)
    
    def __exit__(self, exc_type, exc_value, traceback):
        print(f"關(guān)閉文件: {self.filename}")
        self.file.close()

# 使用跟蹤的文件操作
with TrackedFile('test.txt', 'w') as f:
    f.write('Hello, World!')

leaked_count = tracker.get_leaked_resources()
print(f"泄漏的資源數(shù)量: {leaked_count}")

import json
import yaml
from pathlib import Path

class ConfigManager:
    def __init__(self, config_file):
        self.config_file = Path(config_file)
        self.config = {}
    
    def __enter__(self):
        if self.config_file.exists():
            with open(self.config_file, 'r', encoding='utf-8') as f:
                if self.config_file.suffix.lower() == '.json':
                    self.config = json.load(f)
                elif self.config_file.suffix.lower() in ['.yml', '.yaml']:
                    self.config = yaml.safe_load(f)
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is None:  # 只有在沒有異常時(shí)才保存
            with open(self.config_file, 'w', encoding='utf-8') as f:
                if self.config_file.suffix.lower() == '.json':
                    json.dump(self.config, f, indent=2, ensure_ascii=False)
                elif self.config_file.suffix.lower() in ['.yml', '.yaml']:
                    yaml.dump(self.config, f, default_flow_style=False)
    
    def get(self, key, default=None):
        return self.config.get(key, default)
    
    def set(self, key, value):
        self.config[key] = value

# 使用配置管理器
with ConfigManager('app_config.json') as config:
    config.set('database_url', 'postgresql://localhost/mydb')
    config.set('debug', True)
    db_url = config.get('database_url')
    print(f"數(shù)據(jù)庫(kù)URL: {db_url}")

import datetime
from contextlib import contextmanager

@contextmanager
def log_session(session_name, log_file='app.log'):
    timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # 記錄會(huì)話開始
    with open(log_file, 'a', encoding='utf-8') as f:
        f.write(f"[{timestamp}] 開始會(huì)話: {session_name}\n")
    
    try:
        yield
    except Exception as e:
        # 記錄異常
        error_timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(f"[{error_timestamp}] 異常: {type(e).__name__}: {e}\n")
        raise
    finally:
        # 記錄會(huì)話結(jié)束
        end_timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(f"[{end_timestamp}] 結(jié)束會(huì)話: {session_name}\n")

# 使用日志會(huì)話
with log_session("數(shù)據(jù)處理任務(wù)"):
    print("開始處理數(shù)據(jù)...")
    # 模擬一些工作
    time.sleep(1)
    print("數(shù)據(jù)處理完成")
    # 如果需要測(cè)試異常處理，取消下面這行的注釋
    # raise ValueError("模擬異常")

from functools import wraps
from contextlib import contextmanager

@contextmanager
def performance_monitor(operation_name):
    import time
    start_time = time.time()
    print(f"開始執(zhí)行: {operation_name}")
    try:
        yield
    finally:
        end_time = time.time()
        print(f"{operation_name} 完成，耗時(shí): {end_time - start_time:.2f} 秒")

def monitored(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        with performance_monitor(func.__name__):
            return func(*args, **kwargs)
    return wrapper

@monitored
def slow_function():
    time.sleep(1)
    return "完成"

result = slow_function()

def file_line_generator(filename):
    """逐行讀取文件的生成器"""
    with open(filename, 'r', encoding='utf-8') as f:
        for line_number, line in enumerate(f, 1):
            yield line_number, line.strip()

# 使用生成器處理大文件
def process_large_file_with_generator(filename):
    line_count = 0
    word_count = 0
    
    for line_num, line in file_line_generator(filename):
        line_count += 1
        word_count += len(line.split())
        
        if line_num % 1000 == 0:
            print(f"已處理 {line_num} 行")
    
    return line_count, word_count

# 創(chuàng)建測(cè)試文件
test_lines = [f"這是第{i}行，包含一些測(cè)試數(shù)據(jù)\n" for i in range(1, 10001)]
with open('large_test.txt', 'w', encoding='utf-8') as f:
    f.writelines(test_lines)

# 處理大文件
lines, words = process_large_file_with_generator('large_test.txt')
print(f"總行數(shù): {lines}, 總詞數(shù): {words}")

# 不推薦：過多嵌套
with open('input.txt', 'r') as infile:
    with open('output.txt', 'w') as outfile:
        with open('log.txt', 'a') as logfile:
            # 處理邏輯
            pass

# 推薦：在同一行聲明多個(gè)上下文管理器
with open('input.txt', 'r') as infile, \
     open('output.txt', 'w') as outfile, \
     open('log.txt', 'a') as logfile:
    # 處理邏輯
    pass

# 或者使用contextlib.ExitStack
from contextlib import ExitStack

def process_with_exit_stack():
    with ExitStack() as stack:
        infile = stack.enter_context(open('input.txt', 'r'))
        outfile = stack.enter_context(open('output.txt', 'w'))
        logfile = stack.enter_context(open('log.txt', 'a'))
        
        # 處理邏輯
        pass

from contextlib import nullcontext

def conditional_file_operation(use_log=True):
    # 根據(jù)條件選擇上下文管理器
    log_context = open('operation.log', 'a') if use_log else nullcontext()
    
    with log_context as log_file:
        print("執(zhí)行操作...")
        if log_file:
            log_file.write("操作執(zhí)行成功\n")

conditional_file_operation(True)
conditional_file_operation(False)

import pandas as pd
from contextlib import contextmanager

@contextmanager
def csv_processing_context(csv_file, output_file):
    """CSV處理上下文管理器"""
    print(f"開始處理CSV文件: {csv_file}")
    
    # 讀取數(shù)據(jù)
    df = pd.read_csv(csv_file)
    original_shape = df.shape
    
    try:
        yield df
    finally:
        # 保存處理后的數(shù)據(jù)
        df.to_csv(output_file, index=False)
        print(f"處理完成，原數(shù)據(jù)形狀: {original_shape}，新數(shù)據(jù)形狀: {df.shape}")

# 使用示例
# 創(chuàng)建測(cè)試數(shù)據(jù)
test_data = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35],
    'city': ['New York', 'London', 'Tokyo']
})
test_data.to_csv('test_input.csv', index=False)

# 處理CSV文件
# with csv_processing_context('test_input.csv', 'test_output.csv') as df:
#     df['age_group'] = df['age'].apply(lambda x: 'Young' if x < 30 else 'Adult')
#     print(df.head())

# 假設(shè)使用sqlite3數(shù)據(jù)庫(kù)
import sqlite3
from contextlib import contextmanager

@contextmanager
def database_transaction(db_path):
    """數(shù)據(jù)庫(kù)事務(wù)上下文管理器"""
    conn = sqlite3.connect(db_path)
    try:
        yield conn
        conn.commit()  # 提交事務(wù)
        print("事務(wù)提交成功")
    except Exception as e:
        conn.rollback()  # 回滾事務(wù)
        print(f"事務(wù)回滾: {e}")
        raise
    finally:
        conn.close()

# 使用數(shù)據(jù)庫(kù)事務(wù)
# with database_transaction('example.db') as conn:
#     cursor = conn.cursor()
#     cursor.execute('''CREATE TABLE IF NOT EXISTS users 
#                      (id INTEGER PRIMARY KEY, name TEXT, email TEXT)''')
#     cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", 
#                   ("張三", "zhangsan@example.com"))

from unittest.mock import patch, MagicMock

def test_with_statement():
    """測(cè)試with語(yǔ)句的行為"""
    mock_file = MagicMock()
    mock_file.__enter__.return_value = mock_file
    mock_file.__exit__.return_value = None
    
    with patch('builtins.open', return_value=mock_file):
        with open('test.txt', 'r') as f:
            f.read()
    
    # 驗(yàn)證文件方法被正確調(diào)用
    mock_file.__enter__.assert_called_once()
    mock_file.read.assert_called_once()
    mock_file.__exit__.assert_called_once()

# 運(yùn)行測(cè)試
test_with_statement()
print("測(cè)試通過 ?")

import traceback
from contextlib import contextmanager

@contextmanager
def debug_context(name):
    """調(diào)試用的上下文管理器"""
    print(f"進(jìn)入上下文: {name}")
    print(f"調(diào)用棧:")
    for frame_info in traceback.extract_stack()[:-1]:
        print(f"  {frame_info.filename}:{frame_info.lineno} in {frame_info.name}")
    
    try:
        yield
    except Exception as e:
        print(f"在上下文 {name} 中捕獲異常: {type(e).__name__}: {e}")
        raise
    finally:
        print(f"退出上下文: {name}")

# 使用調(diào)試上下文
with debug_context("主程序"):
    print("執(zhí)行主要邏輯")
    with debug_context("子任務(wù)"):
        print("執(zhí)行子任務(wù)")
        # raise ValueError("測(cè)試異常")  # 取消注釋測(cè)試異常處理

from contextlib import ExitStack

def batch_file_operations(filenames, operation='read'):
    """批量文件操作優(yōu)化"""
    results = []
    
    with ExitStack() as stack:
        files = []
        
        # 打開所有文件
        for filename in filenames:
            if operation == 'read':
                file_obj = stack.enter_context(open(filename, 'r', encoding='utf-8'))
            else:
                file_obj = stack.enter_context(open(filename, 'w', encoding='utf-8'))
            files.append((filename, file_obj))
        
        # 執(zhí)行操作
        for filename, file_obj in files:
            if operation == 'read':
                content = file_obj.read()
                results.append((filename, len(content)))
            else:
                file_obj.write(f"處理文件: {filename}\n")
    
    return results

# 創(chuàng)建測(cè)試文件
for i in range(5):
    with open(f'test_{i}.txt', 'w', encoding='utf-8') as f:
        f.write(f"這是測(cè)試文件 {i} 的內(nèi)容\n" * 10)

# 批量讀取文件
results = batch_file_operations([f'test_{i}.txt' for i in range(5)], 'read')
for filename, size in results:
    print(f"{filename}: {size} 字符")

import mmap

class MappedFileManager:
    def __init__(self, filename, mode='r'):
        self.filename = filename
        self.mode = mode
        self.file = None
        self.mmap_obj = None
    
    def __enter__(self):
        # 打開文件
        self.file = open(self.filename, 'r+b' if 'w' in self.mode else 'rb')
        
        # 創(chuàng)建內(nèi)存映射
        self.mmap_obj = mmap.mmap(
            self.file.fileno(), 
            0, 
            access=mmap.ACCESS_WRITE if 'w' in self.mode else mmap.ACCESS_READ
        )
        
        return self.mmap_obj
    
    def __exit__(self, exc_type, exc_value, traceback):
        if self.mmap_obj:
            self.mmap_obj.close()
        if self.file:
            self.file.close()

# 使用內(nèi)存映射文件
# with MappedFileManager('large_file.bin', 'w') as mm:
#     mm.write(b'Hello, Memory-Mapped World!')

import os
from contextlib import contextmanager

@contextmanager
def secure_file_access(filename, mode='r'):
    """安全的文件訪問上下文管理器"""
    
    # 檢查文件是否存在
    if 'w' in mode and os.path.exists(filename):
        # 檢查寫權(quán)限
        if not os.access(filename, os.W_OK):
            raise PermissionError(f"沒有寫入權(quán)限: {filename}")
    elif 'r' in mode:
        # 檢查讀權(quán)限
        if not os.access(filename, os.R_OK):
            raise PermissionError(f"沒有讀取權(quán)限: {filename}")
    
    # 檢查路徑遍歷攻擊
    if '..' in filename:
        raise ValueError("不允許相對(duì)路徑訪問")
    
    with open(filename, mode, encoding='utf-8') as f:
        yield f

# 安全文件訪問示例
try:
    with secure_file_access('safe_file.txt', 'w') as f:
        f.write('安全的內(nèi)容')
    print("文件寫入成功")
except PermissionError as e:
    print(f"權(quán)限錯(cuò)誤: {e}")
except ValueError as e:
    print(f"安全錯(cuò)誤: {e}")

import tempfile
import os
from contextlib import contextmanager

@contextmanager
def secure_temp_file(suffix='', prefix='tmp'):
    """安全的臨時(shí)文件管理器"""
    temp_file = None
    try:
        # 創(chuàng)建臨時(shí)文件
        temp_file = tempfile.NamedTemporaryFile(
            suffix=suffix, 
            prefix=prefix, 
            delete=False
        )
        temp_filename = temp_file.name
        temp_file.close()
        
        # 設(shè)置安全權(quán)限
        os.chmod(temp_filename, 0o600)  # 只有所有者可讀寫
        
        yield temp_filename
        
    finally:
        # 確保臨時(shí)文件被刪除
        if temp_file and os.path.exists(temp_file.name):
            os.unlink(temp_file.name)

# 使用安全臨時(shí)文件
with secure_temp_file('.txt', 'myapp_') as temp_file:
    with open(temp_file, 'w') as f:
        f.write('臨時(shí)數(shù)據(jù)')
    print(f"臨時(shí)文件路徑: {temp_file}")
    # 文件在這里會(huì)被自動(dòng)刪除

from typing import ContextManager, TextIO
from contextlib import contextmanager

@contextmanager
def typed_file_manager(filename: str, mode: str = 'r') -> ContextManager[TextIO]:
    """帶類型提示的文件管理器"""
    with open(filename, mode, encoding='utf-8') as f:
        yield f

# 使用帶類型的上下文管理器
with typed_file_manager('example.txt', 'w') as f:
    f.write('類型安全的文件操作')

import asyncio
from typing import AsyncContextManager
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_file_manager(filename: str, mode: str = 'r') -> AsyncContextManager:
    """異步文件管理器"""
    loop = asyncio.get_event_loop()
    
    # 在線程池中執(zhí)行阻塞的文件操作
    file_obj = await loop.run_in_executor(None, open, filename, mode)
    
    try:
        yield file_obj
    finally:
        await loop.run_in_executor(None, file_obj.close)

# 使用異步上下文管理器
async def async_file_operation():
    async with async_file_manager('async_example.txt', 'w') as f:
        await asyncio.get_event_loop().run_in_executor(None, f.write, '異步文件內(nèi)容')

# 運(yùn)行異步函數(shù)
# asyncio.run(async_file_operation())