def basic_write_operations():
    """基礎(chǔ)文件寫(xiě)入操作示例"""
    # 模式 'w' - 如果文件不存在則創(chuàng)建，存在則覆蓋
    with open('new_file.txt', 'w', encoding='utf-8') as f:
        f.write("這是新文件的內(nèi)容\n")
        f.write("第二行內(nèi)容\n")
    
    print("'w' 模式寫(xiě)入完成")
    
    # 模式 'a' - 如果文件不存在則創(chuàng)建，存在則追加
    with open('append_file.txt', 'a', encoding='utf-8') as f:
        f.write("追加的內(nèi)容\n")
        f.write("更多追加內(nèi)容\n")
    
    print("'a' 模式寫(xiě)入完成")
    
    # 模式 'x' - 獨(dú)占創(chuàng)建，文件存在則失敗
    try:
        with open('exclusive_file.txt', 'x', encoding='utf-8') as f:
            f.write("獨(dú)占創(chuàng)建的內(nèi)容\n")
        print("'x' 模式寫(xiě)入成功")
    except FileExistsError:
        print("'x' 模式失?。何募汛嬖?)

# 執(zhí)行示例
basic_write_operations()

def safe_write_with_try_except():
    """使用try-except安全處理文件寫(xiě)入"""
    filename = 'data.txt'
    
    try:
        # 嘗試讀取文件檢查是否存在
        with open(filename, 'r', encoding='utf-8') as f:
            content = f.read()
            print("文件已存在，內(nèi)容:")
            print(content)
            
    except FileNotFoundError:
        print("文件不存在，創(chuàng)建新文件")
        
        # 安全寫(xiě)入新文件
        try:
            with open(filename, 'w', encoding='utf-8') as f:
                f.write("新文件的內(nèi)容\n")
                f.write("創(chuàng)建時(shí)間: 2024-01-15\n")
            print("新文件創(chuàng)建成功")
            
        except PermissionError:
            print("錯(cuò)誤: 沒(méi)有寫(xiě)入權(quán)限")
        except Exception as e:
            print(f"寫(xiě)入文件時(shí)發(fā)生錯(cuò)誤: {e}")
            
    except PermissionError:
        print("錯(cuò)誤: 沒(méi)有讀取權(quán)限")
    except Exception as e:
        print(f"檢查文件時(shí)發(fā)生錯(cuò)誤: {e}")

# 執(zhí)行示例
safe_write_with_try_except()

from pathlib import Path

def pathlib_file_operations():
    """使用pathlib進(jìn)行現(xiàn)代文件操作"""
    file_path = Path('data.txt')
    
    # 檢查文件是否存在
    if file_path.exists():
        print("文件已存在")
        print(f"文件大小: {file_path.stat().st_size} 字節(jié)")
        print(f"修改時(shí)間: {file_path.stat().st_mtime}")
        
        # 讀取內(nèi)容
        content = file_path.read_text(encoding='utf-8')
        print("文件內(nèi)容:")
        print(content)
    else:
        print("文件不存在，創(chuàng)建新文件")
        
        # 確保目錄存在
        file_path.parent.mkdir(parents=True, exist_ok=True)
        
        # 寫(xiě)入新文件
        file_path.write_text("使用pathlib創(chuàng)建的內(nèi)容\n第二行\(zhòng)n", encoding='utf-8')
        print("文件創(chuàng)建成功")
    
    # 更復(fù)雜的寫(xiě)入操作
    if not file_path.exists():
        with file_path.open('w', encoding='utf-8') as f:
            f.write("第一行內(nèi)容\n")
            f.write("第二行內(nèi)容\n")
            f.write(f"創(chuàng)建時(shí)間: {datetime.now().isoformat()}\n")
        
        print("使用open方法創(chuàng)建文件")

# 執(zhí)行示例
pathlib_file_operations()

def atomic_write_operations():
    """原子文件寫(xiě)入操作"""
    import tempfile
    import os
    
    def atomic_write(filename, content, encoding='utf-8'):
        """
        原子寫(xiě)入文件
        先寫(xiě)入臨時(shí)文件，然后重命名，避免寫(xiě)入過(guò)程中斷導(dǎo)致文件損壞
        """
        # 創(chuàng)建臨時(shí)文件
        temp_file = None
        try:
            # 在相同目錄創(chuàng)建臨時(shí)文件
            temp_file = tempfile.NamedTemporaryFile(
                mode='w', 
                encoding=encoding,
                dir=os.path.dirname(filename),
                delete=False
            )
            
            # 寫(xiě)入內(nèi)容
            temp_file.write(content)
            temp_file.close()
            
            # 原子替換原文件
            os.replace(temp_file.name, filename)
            print(f"原子寫(xiě)入完成: {filename}")
            
        except Exception as e:
            # 清理臨時(shí)文件
            if temp_file and os.path.exists(temp_file.name):
                os.unlink(temp_file.name)
            raise e
    
    # 使用原子寫(xiě)入
    try:
        content = "原子寫(xiě)入的內(nèi)容\n第二行\(zhòng)n第三行"
        atomic_write('atomic_data.txt', content)
        
    except Exception as e:
        print(f"原子寫(xiě)入失敗: {e}")
    
    # 帶備份的原子寫(xiě)入
    def atomic_write_with_backup(filename, content, encoding='utf-8'):
        """帶備份的原子寫(xiě)入"""
        backup_file = filename + '.bak'
        
        # 如果原文件存在，創(chuàng)建備份
        if os.path.exists(filename):
            os.replace(filename, backup_file)
        
        try:
            atomic_write(filename, content, encoding)
            # 成功后刪除備份
            if os.path.exists(backup_file):
                os.unlink(backup_file)
                
        except Exception as e:
            # 失敗時(shí)恢復(fù)備份
            if os.path.exists(backup_file):
                os.replace(backup_file, filename)
            raise e
    
    # 測(cè)試帶備份的寫(xiě)入
    try:
        content = "帶備份的原子寫(xiě)入內(nèi)容"
        atomic_write_with_backup('backup_data.txt', content)
        print("帶備份的原子寫(xiě)入完成")
        
    except Exception as e:
        print(f"帶備份的原子寫(xiě)入失敗: {e}")

# 執(zhí)行示例
atomic_write_operations()

def file_locking_mechanism():
    """文件鎖機(jī)制實(shí)現(xiàn)"""
    import fcntl  # Unix系統(tǒng)文件鎖
    import time
    
    def write_with_lock(filename, content, timeout=10):
        """使用文件鎖進(jìn)行安全寫(xiě)入"""
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            try:
                with open(filename, 'a+', encoding='utf-8') as f:  # 使用a+模式創(chuàng)建文件
                    # 獲取獨(dú)占鎖
                    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
                    
                    # 檢查文件內(nèi)容（如果需要）
                    f.seek(0)
                    existing_content = f.read()
                    
                    # 寫(xiě)入新內(nèi)容
                    f.seek(0)
                    f.truncate()  # 清空文件
                    f.write(content)
                    
                    # 釋放鎖會(huì)在文件關(guān)閉時(shí)自動(dòng)進(jìn)行
                    print(f"寫(xiě)入完成: {filename}")
                    return True
                    
            except BlockingIOError:
                # 文件被鎖定，等待重試
                print("文件被鎖定，等待...")
                time.sleep(0.1)
            except FileNotFoundError:
                # 文件不存在，嘗試創(chuàng)建
                try:
                    with open(filename, 'x', encoding='utf-8') as f:
                        f.write(content)
                    print(f"創(chuàng)建并寫(xiě)入: {filename}")
                    return True
                except FileExistsError:
                    # 在創(chuàng)建過(guò)程中文件被其他進(jìn)程創(chuàng)建
                    continue
            except Exception as e:
                print(f"寫(xiě)入錯(cuò)誤: {e}")
                return False
        
        print("獲取文件鎖超時(shí)")
        return False
    
    # 測(cè)試文件鎖
    success = write_with_lock('locked_file.txt', "使用文件鎖寫(xiě)入的內(nèi)容\n")
    print(f"寫(xiě)入結(jié)果: {'成功' if success else '失敗'}")
    
    # Windows平臺(tái)替代方案
    def windows_file_lock(filename, content):
        """Windows平臺(tái)文件鎖模擬"""
        try:
            # 使用獨(dú)占模式打開(kāi)文件
            with open(filename, 'w', encoding='utf-8') as f:
                # 在Windows上，簡(jiǎn)單的打開(kāi)操作就提供了基本的獨(dú)占性
                f.write(content)
                print("Windows文件寫(xiě)入完成")
                return True
                
        except PermissionError:
            print("文件被其他進(jìn)程鎖定")
            return False
        except Exception as e:
            print(f"寫(xiě)入錯(cuò)誤: {e}")
            return False
    
    # 根據(jù)平臺(tái)選擇方法
    if os.name == 'posix':  # Unix/Linux/Mac
        write_with_lock('unix_file.txt', "Unix系統(tǒng)內(nèi)容\n")
    elif os.name == 'nt':   # Windows
        windows_file_lock('windows_file.txt', "Windows系統(tǒng)內(nèi)容\n")

# 執(zhí)行示例
file_locking_mechanism()

def multiprocess_file_safety():
    """多進(jìn)程環(huán)境下的文件安全寫(xiě)入"""
    import multiprocessing
    import time
    
    def worker_process(process_id, lock, output_file):
        """工作進(jìn)程函數(shù)"""
        for i in range(5):
            # 模擬一些處理
            time.sleep(0.1)
            
            # 獲取鎖保護(hù)文件寫(xiě)入
            with lock:
                try:
                    with open(output_file, 'a', encoding='utf-8') as f:
                        timestamp = time.time()
                        message = f"進(jìn)程 {process_id} - 迭代 {i} - 時(shí)間 {timestamp:.6f}\n"
                        f.write(message)
                        print(f"進(jìn)程 {process_id} 寫(xiě)入完成")
                        
                except Exception as e:
                    print(f"進(jìn)程 {process_id} 寫(xiě)入錯(cuò)誤: {e}")
    
    # 創(chuàng)建進(jìn)程鎖
    manager = multiprocessing.Manager()
    lock = manager.Lock()
    
    output_file = 'multiprocess_output.txt'
    
    # 清理舊文件（如果存在）
    if os.path.exists(output_file):
        os.remove(output_file)
    
    # 創(chuàng)建多個(gè)進(jìn)程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(
            target=worker_process,
            args=(i, lock, output_file)
        )
        processes.append(p)
        p.start()
    
    # 等待所有進(jìn)程完成
    for p in processes:
        p.join()
    
    print("所有進(jìn)程完成")
    
    # 顯示結(jié)果
    if os.path.exists(output_file):
        with open(output_file, 'r', encoding='utf-8') as f:
            content = f.read()
            print("最終文件內(nèi)容:")
            print(content)
    
    # 統(tǒng)計(jì)結(jié)果
    if os.path.exists(output_file):
        with open(output_file, 'r', encoding='utf-8') as f:
            lines = f.readlines()
            print(f"總行數(shù): {len(lines)}")
            
            # 按進(jìn)程統(tǒng)計(jì)
            process_counts = {}
            for line in lines:
                if line.startswith('進(jìn)程'):
                    parts = line.split(' - ')
                    process_id = int(parts[0].split()[1])
                    process_counts[process_id] = process_counts.get(process_id, 0) + 1
            
            print("各進(jìn)程寫(xiě)入行數(shù):", process_counts)

# 執(zhí)行示例
multiprocess_file_safety()

def multithreaded_file_safety():
    """多線程環(huán)境下的文件安全寫(xiě)入"""
    import threading
    import queue
    import time
    
    class ThreadSafeFileWriter:
        """線程安全的文件寫(xiě)入器"""
        def __init__(self, filename, max_queue_size=1000):
            self.filename = filename
            self.write_queue = queue.Queue(max_queue_size)
            self.stop_event = threading.Event()
            self.worker_thread = None
            
        def start(self):
            """啟動(dòng)寫(xiě)入線程"""
            self.worker_thread = threading.Thread(target=self._write_worker)
            self.worker_thread.daemon = True
            self.worker_thread.start()
        
        def stop(self):
            """停止寫(xiě)入線程"""
            self.stop_event.set()
            if self.worker_thread:
                self.worker_thread.join()
        
        def write(self, message):
            """寫(xiě)入消息（線程安全）"""
            try:
                self.write_queue.put(message, timeout=1)
                return True
            except queue.Full:
                print("寫(xiě)入隊(duì)列已滿")
                return False
        
        def _write_worker(self):
            """寫(xiě)入工作線程"""
            # 確保文件存在
            if not os.path.exists(self.filename):
                with open(self.filename, 'w', encoding='utf-8') as f:
                    f.write("文件頭\n")
            
            while not self.stop_event.is_set() or not self.write_queue.empty():
                try:
                    # 獲取消息
                    message = self.write_queue.get(timeout=0.1)
                    
                    # 寫(xiě)入文件
                    with open(self.filename, 'a', encoding='utf-8') as f:
                        f.write(message + '\n')
                    
                    self.write_queue.task_done()
                    
                except queue.Empty:
                    continue
                except Exception as e:
                    print(f"寫(xiě)入錯(cuò)誤: {e}")
    
    def writer_thread(thread_id, writer):
        """寫(xiě)入線程函數(shù)"""
        for i in range(10):
            message = f"線程 {thread_id} - 消息 {i} - 時(shí)間 {time.time():.6f}"
            success = writer.write(message)
            if not success:
                print(f"線程 {thread_id} 寫(xiě)入失敗")
            time.sleep(0.05)
    
    # 使用線程安全寫(xiě)入器
    output_file = 'threadsafe_output.txt'
    
    # 清理舊文件
    if os.path.exists(output_file):
        os.remove(output_file)
    
    writer = ThreadSafeFileWriter(output_file)
    writer.start()
    
    try:
        # 創(chuàng)建多個(gè)寫(xiě)入線程
        threads = []
        for i in range(3):
            t = threading.Thread(target=writer_thread, args=(i, writer))
            threads.append(t)
            t.start()
        
        # 等待寫(xiě)入線程完成
        for t in threads:
            t.join()
        
        # 等待隊(duì)列清空
        writer.write_queue.join()
        
    finally:
        writer.stop()
    
    print("所有線程完成")
    
    # 顯示結(jié)果
    if os.path.exists(output_file):
        with open(output_file, 'r', encoding='utf-8') as f:
            content = f.read()
            print("最終文件內(nèi)容:")
            print(content)
            
            # 統(tǒng)計(jì)行數(shù)
            lines = content.split('\n')
            print(f"總行數(shù): {len([l for l in lines if l.strip()])}")

# 執(zhí)行示例
multithreaded_file_safety()

def logging_system_implementation():
    """完整的日志系統(tǒng)實(shí)現(xiàn)"""
    import logging
    from logging.handlers import RotatingFileHandler
    from datetime import datetime
    
    class SafeRotatingFileHandler(RotatingFileHandler):
        """安全的循環(huán)文件處理器"""
        def __init__(self, filename, **kwargs):
            # 確保目錄存在
            os.makedirs(os.path.dirname(os.path.abspath(filename)), exist_ok=True)
            super().__init__(filename, **kwargs)
        
        def _open(self):
            """安全打開(kāi)文件"""
            try:
                return super()._open()
            except FileNotFoundError:
                # 創(chuàng)建文件
                with open(self.baseFilename, 'w', encoding='utf-8'):
                    pass
                return super()._open()
    
    def setup_logging_system():
        """設(shè)置日志系統(tǒng)"""
        # 創(chuàng)建日志目錄
        log_dir = 'logs'
        os.makedirs(log_dir, exist_ok=True)
        
        # 主日志文件
        main_log = os.path.join(log_dir, 'application.log')
        
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                SafeRotatingFileHandler(
                    main_log,
                    maxBytes=1024 * 1024,  # 1MB
                    backupCount=5,
                    encoding='utf-8'
                ),
                logging.StreamHandler()  # 同時(shí)輸出到控制臺(tái)
            ]
        )
        
        return logging.getLogger('Application')
    
    def test_logging_system():
        """測(cè)試日志系統(tǒng)"""
        logger = setup_logging_system()
        
        # 記錄不同級(jí)別的消息
        logger.debug("調(diào)試信息")
        logger.info("一般信息")
        logger.warning("警告信息")
        logger.error("錯(cuò)誤信息")
        logger.critical("嚴(yán)重錯(cuò)誤")
        
        # 記錄帶上下文的信息
        try:
            # 模擬錯(cuò)誤
            result = 1 / 0
        except Exception as e:
            logger.exception("除零錯(cuò)誤發(fā)生")
        
        # 性能日志
        start_time = time.time()
        time.sleep(0.1)  # 模擬工作
        end_time = time.time()
        logger.info(f"操作完成，耗時(shí): {(end_time - start_time)*1000:.2f}ms")
    
    # 運(yùn)行測(cè)試
    test_logging_system()
    print("日志系統(tǒng)測(cè)試完成")
    
    # 檢查日志文件
    log_file = 'logs/application.log'
    if os.path.exists(log_file):
        with open(log_file, 'r', encoding='utf-8') as f:
            lines = f.readlines()
            print(f"日志文件行數(shù): {len(lines)}")
            print("最后5行日志:")
            for line in lines[-5:]:
                print(line.strip())

# 執(zhí)行示例
logging_system_implementation()

def data_export_system():
    """安全的數(shù)據(jù)導(dǎo)出系統(tǒng)"""
    import csv
    import json
    from datetime import datetime
    
    class DataExporter:
        """數(shù)據(jù)導(dǎo)出器"""
        def __init__(self, output_dir='exports'):
            self.output_dir = output_dir
            os.makedirs(output_dir, exist_ok=True)
        
        def export_csv(self, data, filename, headers=None):
            """導(dǎo)出CSV文件"""
            filepath = os.path.join(self.output_dir, filename)
            
            # 安全寫(xiě)入CSV
            try:
                with open(filepath, 'x', encoding='utf-8', newline='') as f:
                    writer = csv.writer(f)
                    
                    if headers:
                        writer.writerow(headers)
                    
                    for row in data:
                        writer.writerow(row)
                
                print(f"CSV導(dǎo)出成功: {filepath}")
                return True
                
            except FileExistsError:
                print(f"文件已存在: {filepath}")
                return False
            except Exception as e:
                print(f"CSV導(dǎo)出失敗: {e}")
                return False
        
        def export_json(self, data, filename):
            """導(dǎo)出JSON文件"""
            filepath = os.path.join(self.output_dir, filename)
            
            # 使用原子寫(xiě)入
            try:
                # 先寫(xiě)入臨時(shí)文件
                temp_file = filepath + '.tmp'
                with open(temp_file, 'x', encoding='utf-8') as f:
                    json.dump(data, f, indent=2, ensure_ascii=False)
                
                # 原子替換
                os.replace(temp_file, filepath)
                
                print(f"JSON導(dǎo)出成功: {filepath}")
                return True
                
            except FileExistsError:
                print(f"文件已存在: {filepath}")
                return False
            except Exception as e:
                # 清理臨時(shí)文件
                if os.path.exists(temp_file):
                    os.remove(temp_file)
                print(f"JSON導(dǎo)出失敗: {e}")
                return False
        
        def export_with_timestamp(self, data, base_filename, format='csv'):
            """帶時(shí)間戳的導(dǎo)出"""
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            filename = f"{base_filename}_{timestamp}.{format}"
            
            if format == 'csv':
                return self.export_csv(data, filename)
            elif format == 'json':
                return self.export_json(data, filename)
            else:
                print(f"不支持的格式: {format}")
                return False
    
    # 使用數(shù)據(jù)導(dǎo)出器
    exporter = DataExporter('data_exports')
    
    # 示例數(shù)據(jù)
    sample_data = [
        ['姓名', '年齡', '城市'],
        ['張三', 25, '北京'],
        ['李四', 30, '上海'],
        ['王五', 28, '廣州']
    ]
    
    # 導(dǎo)出CSV
    success = exporter.export_csv(
        sample_data[1:],  # 數(shù)據(jù)行
        'users.csv',
        headers=sample_data[0]  # 表頭
    )
    
    # 導(dǎo)出JSON
    json_data = [
        {'name': '張三', 'age': 25, 'city': '北京'},
        {'name': '李四', 'age': 30, 'city': '上海'},
        {'name': '王五', 'age': 28, 'city': '廣州'}
    ]
    
    success = exporter.export_json(json_data, 'users.json')
    
    # 帶時(shí)間戳導(dǎo)出
    success = exporter.export_with_timestamp(
        sample_data[1:],
        'users_backup',
        format='csv'
    )
    
    # 列出導(dǎo)出文件
    print("導(dǎo)出文件列表:")
    for file in os.listdir('data_exports'):
        if file.endswith(('.csv', '.json')):
            filepath = os.path.join('data_exports', file)
            size = os.path.getsize(filepath)
            print(f"  {file}: {size} 字節(jié)")

# 執(zhí)行示例
data_export_system()

def comprehensive_error_handling():
    """完整的文件操作錯(cuò)誤處理"""
    class FileOperationError(Exception):
        """文件操作錯(cuò)誤基類"""
        pass
    
    class FileExistsError(FileOperationError):
        """文件已存在錯(cuò)誤"""
        pass
    
    class FilePermissionError(FileOperationError):
        """文件權(quán)限錯(cuò)誤"""
        pass
    
    class FileSystemError(FileOperationError):
        """文件系統(tǒng)錯(cuò)誤"""
        pass
    
    class SafeFileWriter:
        """安全的文件寫(xiě)入器"""
        def __init__(self, max_retries=3, retry_delay=0.1):
            self.max_retries = max_retries
            self.retry_delay = retry_delay
        
        def write_file(self, filename, content, mode='x', encoding='utf-8'):
            """安全寫(xiě)入文件"""
            for attempt in range(self.max_retries):
                try:
                    with open(filename, mode, encoding=encoding) as f:
                        f.write(content)
                    return True
                    
                except FileExistsError:
                    raise FileExistsError(f"文件已存在: {filename}")
                    
                except PermissionError:
                    if attempt == self.max_retries - 1:
                        raise FilePermissionError(f"沒(méi)有寫(xiě)入權(quán)限: {filename}")
                    time.sleep(self.retry_delay)
                    
                except OSError as e:
                    if attempt == self.max_retries - 1:
                        raise FileSystemError(f"文件系統(tǒng)錯(cuò)誤: {e}")
                    time.sleep(self.retry_delay)
                    
                except Exception as e:
                    if attempt == self.max_retries - 1:
                        raise FileOperationError(f"未知錯(cuò)誤: {e}")
                    time.sleep(self.retry_delay)
            
            return False
        
        def write_file_with_fallback(self, filename, content, **kwargs):
            """帶回退機(jī)制的寫(xiě)入"""
            try:
                return self.write_file(filename, content, **kwargs)
                
            except FileExistsError:
                # 生成帶時(shí)間戳的回退文件名
                timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
                base_name, ext = os.path.splitext(filename)
                fallback_filename = f"{base_name}_{timestamp}{ext}"
                
                print(f"文件已存在，使用回退文件名: {fallback_filename}")
                return self.write_file(fallback_filename, content, **kwargs)
    
    # 使用安全寫(xiě)入器
    writer = SafeFileWriter(max_retries=3)
    
    # 測(cè)試各種場(chǎng)景
    test_cases = [
        ('test_file.txt', '測(cè)試內(nèi)容', 'x'),  # 正常創(chuàng)建
        ('test_file.txt', '重復(fù)內(nèi)容', 'x'),  # 文件已存在
        ('/root/protected.txt', '權(quán)限測(cè)試', 'x'),  # 權(quán)限錯(cuò)誤（Unix）
    ]
    
    for filename, content, mode in test_cases:
        try:
            print(f"\n嘗試寫(xiě)入: {filename}")
            success = writer.write_file(filename, content, mode)
            if success:
                print("寫(xiě)入成功")
            else:
                print("寫(xiě)入失敗")
                
        except FileExistsError as e:
            print(f"文件存在錯(cuò)誤: {e}")
            # 嘗試使用回退機(jī)制
            try:
                success = writer.write_file_with_fallback(filename, content, mode=mode)
                if success:
                    print("回退寫(xiě)入成功")
            except Exception as fallback_error:
                print(f"回退也失敗: {fallback_error}")
                
        except FilePermissionError as e:
            print(f"權(quán)限錯(cuò)誤: {e}")
            
        except FileSystemError as e:
            print(f"系統(tǒng)錯(cuò)誤: {e}")
            
        except FileOperationError as e:
            print(f"操作錯(cuò)誤: {e}")
            
        except Exception as e:
            print(f"未知錯(cuò)誤: {e}")

# 執(zhí)行示例
comprehensive_error_handling()

def file_operation_monitoring():
    """文件操作監(jiān)控與回滾"""
    class FileOperationMonitor:
        """文件操作監(jiān)控器"""
        def __init__(self):
            self.operations = []  # 記錄所有文件操作
            self.backup_files = {}  # 備份文件映射
        
        def backup_file(self, filename):
            """備份文件"""
            if os.path.exists(filename):
                backup_path = filename + '.backup'
                shutil.copy2(filename, backup_path)
                self.backup_files[filename] = backup_path
                return backup_path
            return None
        
        def record_operation(self, op_type, filename, **kwargs):
            """記錄文件操作"""
            operation = {
                'type': op_type,
                'filename': filename,
                'timestamp': time.time(),
                'kwargs': kwargs
            }
            self.operations.append(operation)
        
        def safe_write(self, filename, content, **kwargs):
            """安全寫(xiě)入文件"""
            # 備份原文件
            backup_path = self.backup_file(filename)
            
            try:
                # 記錄操作
                self.record_operation('write', filename, content=content[:100])
                
                # 執(zhí)行寫(xiě)入
                with open(filename, 'w', encoding='utf-8') as f:
                    f.write(content)
                
                return True
                
            except Exception as e:
                # 回滾：恢復(fù)備份
                if backup_path and os.path.exists(backup_path):
                    shutil.copy2(backup_path, filename)
                    print(f"回滾: 恢復(fù) {filename} 從備份")
                
                # 清理備份
                if backup_path and os.path.exists(backup_path):
                    os.remove(backup_path)
                
                raise e
        
        def rollback(self):
            """回滾所有操作"""
            print("執(zhí)行回滾操作...")
            success_count = 0
            
            for operation in reversed(self.operations):
                try:
                    if operation['type'] == 'write':
                        filename = operation['filename']
                        if filename in self.backup_files:
                            backup_path = self.backup_files[filename]
                            if os.path.exists(backup_path):
                                shutil.copy2(backup_path, filename)
                                success_count += 1
                                print(f"回滾寫(xiě)入: {filename}")
                    
                    # 可以添加其他操作類型的回滾邏輯
                    
                except Exception as e:
                    print(f"回滾失敗 {operation['type']} {operation['filename']}: {e}")
            
            # 清理所有備份
            for backup_path in self.backup_files.values():
                if os.path.exists(backup_path):
                    try:
                        os.remove(backup_path)
                    except:
                        pass
            
            print(f"回滾完成，成功恢復(fù) {success_count} 個(gè)文件")
    
    # 使用監(jiān)控器
    monitor = FileOperationMonitor()
    
    try:
        # 執(zhí)行一系列文件操作
        test_files = ['test1.txt', 'test2.txt', 'test3.txt']
        
        for i, filename in enumerate(test_files):
            content = f"測(cè)試文件 {i+1} 的內(nèi)容\n" * 10
            
            # 備份并寫(xiě)入
            monitor.safe_write(filename, content)
            print(f"寫(xiě)入成功: {filename}")
        
        # 模擬錯(cuò)誤，觸發(fā)回滾
        raise Exception("模擬的業(yè)務(wù)邏輯錯(cuò)誤")
        
    except Exception as e:
        print(f"發(fā)生錯(cuò)誤: {e}")
        monitor.rollback()
    
    finally:
        # 清理測(cè)試文件
        for filename in test_files:
            if os.path.exists(filename):
                os.remove(filename)
                print(f"清理: {filename}")

# 執(zhí)行示例
file_operation_monitoring()

def professional_file_writer():
    """
    專業(yè)文件寫(xiě)入模板
    包含錯(cuò)誤處理、重試機(jī)制、并發(fā)安全等最佳實(shí)踐
    """
    class ProfessionalFileWriter:
        def __init__(self, max_retries=3, retry_delay=0.1):
            self.max_retries = max_retries
            self.retry_delay = retry_delay
        
        def write_file(self, filename, content, mode='x', encoding='utf-8'):
            """安全寫(xiě)入文件"""
            for attempt in range(self.max_retries):
                try:
                    # 確保目錄存在
                    os.makedirs(os.path.dirname(os.path.abspath(filename)), exist_ok=True)
                    
                    with open(filename, mode, encoding=encoding) as f:
                        f.write(content)
                    
                    # 驗(yàn)證寫(xiě)入結(jié)果
                    if self._verify_write(filename, content, encoding):
                        return True
                    else:
                        raise FileSystemError("寫(xiě)入驗(yàn)證失敗")
                        
                except FileExistsError:
                    if mode == 'x':  # 獨(dú)占模式不允許文件存在
                        raise
                    # 其他模式可以繼續(xù)嘗試
                    time.sleep(self.retry_delay)
                    
                except (PermissionError, OSError) as e:
                    if attempt == self.max_retries - 1:
                        raise
                    time.sleep(self.retry_delay)
                    
                except Exception as e:
                    if attempt == self.max_retries - 1:
                        raise
                    time.sleep(self.retry_delay)
            
            return False
        
        def _verify_write(self, filename, expected_content, encoding='utf-8'):
            """驗(yàn)證寫(xiě)入結(jié)果"""
            try:
                with open(filename, 'r', encoding=encoding) as f:
                    actual_content = f.read()
                return actual_content == expected_content
            except:
                return False
        
        def atomic_write(self, filename, content, encoding='utf-8'):
            """原子寫(xiě)入"""
            import tempfile
            
            # 創(chuàng)建臨時(shí)文件
            temp_file = tempfile.NamedTemporaryFile(
                mode='w', delete=False, encoding=encoding
            )
            
            try:
                # 寫(xiě)入臨時(shí)文件
                with temp_file:
                    temp_file.write(content)
                
                # 原子替換
                os.replace(temp_file.name, filename)
                return True
                
            except Exception as e:
                # 清理臨時(shí)文件
                if os.path.exists(temp_file.name):
                    os.remove(temp_file.name)
                raise e
            
            finally:
                # 確保臨時(shí)文件被清理
                if os.path.exists(temp_file.name):
                    try:
                        os.remove(temp_file.name)
                    except:
                        pass
    
    # 使用示例
    writer = ProfessionalFileWriter()
    
    try:
        success = writer.write_file(
            'professional_output.txt',
            '專業(yè)寫(xiě)入的內(nèi)容\n第二行\(zhòng)n',
            mode='x'
        )
        print(f"寫(xiě)入結(jié)果: {'成功' if success else '失敗'}")
        
    except Exception as e:
        print(f"寫(xiě)入失敗: {e}")
        
        # 嘗試原子寫(xiě)入作為回退
        try:
            success = writer.atomic_write(
                'professional_output.txt',
                '原子寫(xiě)入的內(nèi)容\n'
            )
            print(f"原子寫(xiě)入結(jié)果: {'成功' if success else '失敗'}")
        except Exception as atomic_error:
            print(f"原子寫(xiě)入也失敗: {atomic_error}")

# 執(zhí)行示例
professional_file_writer()