def demonstrate_tempfile_needs():
    """
    演示臨時文件的常見應(yīng)用場景
    """
    scenarios = [
        {
            'name': '數(shù)據(jù)處理中間存儲',
            'description': '大型數(shù)據(jù)集處理過程中的臨時存儲',
            'example': '排序、轉(zhuǎn)換或聚合大量數(shù)據(jù)時的中間文件'
        },
        {
            'name': '緩存系統(tǒng)',
            'description': '存儲計算密集型操作的緩存結(jié)果',
            'example': '網(wǎng)頁緩存、API響應(yīng)緩存或渲染結(jié)果緩存'
        },
        {
            'name': '文件下載和處理',
            'description': '下載文件后的臨時處理空間',
            'example': '下載壓縮包后的解壓和內(nèi)容提取'
        },
        {
            'name': '測試和調(diào)試',
            'description': '單元測試和調(diào)試時創(chuàng)建臨時測試數(shù)據(jù)',
            'example': '自動化測試中的臨時數(shù)據(jù)庫或配置文件'
        },
        {
            'name': '進程間通信',
            'description': '不同進程之間的數(shù)據(jù)交換',
            'example': '多進程應(yīng)用中的共享臨時數(shù)據(jù)存儲'
        }
    ]
    
    print("=== 臨時文件的應(yīng)用場景 ===")
    for scenario in scenarios:
        print(f"\n{scenario['name']}:")
        print(f"  描述: {scenario['description']}")
        print(f"  示例: {scenario['example']}")

demonstrate_tempfile_needs()

def demonstrate_tempfile_challenges():
    """
    演示臨時文件處理中的挑戰(zhàn)
    """
    challenges = [
        {
            'issue': '安全風(fēng)險',
            'description': '臨時文件可能包含敏感數(shù)據(jù)， improper handling can lead to data leaks',
            'solution': '使用安全權(quán)限和加密存儲'
        },
        {
            'issue': '資源泄漏',
            'description': '未正確清理臨時文件會導(dǎo)致磁盤空間耗盡',
            'solution': '自動清理機制和資源管理'
        },
        {
            'issue': '并發(fā)訪問',
            'description': '多進程/線程同時訪問臨時文件可能導(dǎo)致沖突',
            'solution': '使用文件鎖和唯一命名'
        },
        {
            'issue': '跨平臺兼容性',
            'description': '不同操作系統(tǒng)的文件系統(tǒng)特性差異',
            'solution': '使用跨平臺API和路徑處理'
        },
        {
            'issue': '性能考慮',
            'description': '頻繁的文件IO操作可能影響性能',
            'solution': '適當?shù)木彌_和內(nèi)存映射策略'
        }
    ]
    
    print("=== 臨時文件處理的挑戰(zhàn)與解決方案 ===")
    for challenge in challenges:
        print(f"\n{challenge['issue']}:")
        print(f"  問題: {challenge['description']}")
        print(f"  解決方案: {challenge['solution']}")

demonstrate_tempfile_challenges()

import tempfile
import os

def demonstrate_basic_tempfiles():
    """
    演示基本的臨時文件創(chuàng)建方法
    """
    print("=== 基本臨時文件創(chuàng)建 ===")
    
    # 方法1: 使用TemporaryFile - 自動清理
    with tempfile.TemporaryFile(mode='w+', encoding='utf-8') as temp_file:
        temp_file.write("這是臨時文件內(nèi)容\n第二行內(nèi)容")
        temp_file.seek(0)
        content = temp_file.read()
        print(f"TemporaryFile 內(nèi)容: {content!r}")
        print(f"文件描述符: {temp_file.fileno()}")
    
    # 文件在此處已自動刪除
    print("文件已自動刪除")
    
    # 方法2: 使用NamedTemporaryFile - 有名稱的臨時文件
    with tempfile.NamedTemporaryFile(mode='w+', delete=True, encoding='utf-8') as named_temp:
        named_temp.write("命名臨時文件內(nèi)容")
        print(f"臨時文件路徑: {named_temp.name}")
        
        # 驗證文件存在
        if os.path.exists(named_temp.name):
            print("文件存在，可被其他進程訪問")
        else:
            print("文件不存在")
    
    # 方法3: 使用mkstemp - 低級API
    fd, temp_path = tempfile.mkstemp(suffix='.txt', prefix='python_')
    print(f"mkstemp 文件描述符: {fd}, 路徑: {temp_path}")
    
    # 需要手動管理資源
    try:
        with os.fdopen(fd, 'w') as f:
            f.write("mkstemp 創(chuàng)建的內(nèi)容")
        
        # 讀取內(nèi)容驗證
        with open(temp_path, 'r') as f:
            content = f.read()
            print(f"mkstemp 文件內(nèi)容: {content!r}")
    finally:
        # 手動清理
        os.unlink(temp_path)
        print("mkstemp 文件已手動刪除")

demonstrate_basic_tempfiles()

def demonstrate_temp_directories():
    """
    演示臨時目錄的創(chuàng)建和使用
    """
    print("=== 臨時目錄創(chuàng)建與使用 ===")
    
    # 使用TemporaryDirectory
    with tempfile.TemporaryDirectory(prefix='python_temp_', suffix='_dir') as temp_dir:
        print(f"臨時目錄路徑: {temp_dir}")
        
        # 在臨時目錄中創(chuàng)建多個文件
        files_to_create = ['config.ini', 'data.json', 'log.txt']
        
        for filename in files_to_create:
            file_path = os.path.join(temp_dir, filename)
            with open(file_path, 'w') as f:
                f.write(f"{filename} 的內(nèi)容")
            
            print(f"創(chuàng)建文件: {file_path}")
        
        # 驗證文件存在
        created_files = os.listdir(temp_dir)
        print(f"目錄中的文件: {created_files}")
        
        # 演示目錄自動清理
        print("退出with塊后目錄將自動刪除")
    
    # 使用mkdtemp - 手動管理
    temp_dir_path = tempfile.mkdtemp(prefix='manual_')
    print(f"手動管理臨時目錄: {temp_dir_path}")
    
    try:
        # 在目錄中工作
        test_file = os.path.join(temp_dir_path, 'test.txt')
        with open(test_file, 'w') as f:
            f.write("測試內(nèi)容")
        
        print(f"創(chuàng)建測試文件: {test_file}")
        
    finally:
        # 手動清理 - 需要遞歸刪除
        import shutil
        shutil.rmtree(temp_dir_path)
        print("手動臨時目錄已刪除")

demonstrate_temp_directories()

def demonstrate_secure_tempfiles():
    """
    演示安全臨時文件處理
    """
    print("=== 安全臨時文件處理 ===")
    
    # 1. 安全權(quán)限設(shè)置
    with tempfile.NamedTemporaryFile(mode='w', delete=False) as secure_temp:
        # 設(shè)置安全文件權(quán)限 (僅當前用戶可讀寫)
        os.chmod(secure_temp.name, 0o600)
        secure_temp.write("敏感數(shù)據(jù): 密碼、密鑰等")
        print(f"安全文件創(chuàng)建: {secure_temp.name}")
        
        # 驗證權(quán)限
        stat_info = os.stat(secure_temp.name)
        print(f"文件權(quán)限: {oct(stat_info.st_mode)[-3:]}")
    
    # 手動清理
    os.unlink(secure_temp.name)
    
    # 2. 安全目錄位置
    secure_dir = tempfile.mkdtemp()
    try:
        # 在安全目錄中創(chuàng)建文件
        secure_file_path = os.path.join(secure_dir, 'secure_data.bin')
        
        # 使用安全模式創(chuàng)建文件
        with open(secure_file_path, 'wb') as f:
            f.write(b'加密的敏感數(shù)據(jù)')
            os.chmod(secure_file_path, 0o600)
        
        print(f"安全目錄中的文件: {secure_file_path}")
        
    finally:
        shutil.rmtree(secure_dir)
    
    # 3. 使用加密的臨時文件
    from cryptography.fernet import Fernet
    
    # 生成加密密鑰
    key = Fernet.generate_key()
    cipher = Fernet(key)
    
    with tempfile.NamedTemporaryFile(mode='wb', delete=False) as encrypted_temp:
        sensitive_data = "超級機密信息".encode('utf-8')
        encrypted_data = cipher.encrypt(sensitive_data)
        encrypted_temp.write(encrypted_data)
        
        print(f"加密文件創(chuàng)建: {encrypted_temp.name}")
        print(f"原始數(shù)據(jù)長度: {len(sensitive_data)}")
        print(f"加密數(shù)據(jù)長度: {len(encrypted_data)}")
    
    # 清理
    os.unlink(encrypted_temp.name)

demonstrate_secure_tempfiles()

def demonstrate_advanced_tempfile_modes():
    """
    演示高級臨時文件模式和使用技巧
    """
    print("=== 高級臨時文件模式 ===")
    
    # 1. 使用不同的模式組合
    modes = [
        ('w+b', "二進制讀寫"),
        ('w+', "文本讀寫"),
        ('xb', "獨占創(chuàng)建二進制"),
        ('x', "獨占創(chuàng)建文本")
    ]
    
    for mode, description in modes:
        try:
            with tempfile.NamedTemporaryFile(mode=mode, delete=False) as temp_file:
                print(f"模式 {mode} ({description}): {temp_file.name}")
                
                if 'b' in mode:
                    # 二進制操作
                    temp_file.write(b'binary data\x00\x01\x02')
                    temp_file.seek(0)
                    data = temp_file.read()
                    print(f"  寫入/讀取: {data!r}")
                else:
                    # 文本操作
                    temp_file.write("文本數(shù)據(jù)\n多行內(nèi)容")
                    temp_file.seek(0)
                    content = temp_file.read()
                    print(f"  寫入/讀取: {content!r}")
                
            os.unlink(temp_file.name)
            
        except Exception as e:
            print(f"模式 {mode} 錯誤: {e}")
    
    # 2. 自定義前綴和后綴
    custom_temp = tempfile.NamedTemporaryFile(
        prefix='custom_',
        suffix='.json',
        delete=False
    )
    print(f"\n自定義臨時文件: {custom_temp.name}")
    custom_temp.close()
    os.unlink(custom_temp.name)
    
    # 3. 指定目錄和忽略清理
    specific_dir = '/tmp' if os.name != 'nt' else tempfile.gettempdir()
    with tempfile.NamedTemporaryFile(
        dir=specific_dir,
        delete=False  # 不自動刪除，用于演示
    ) as specific_temp:
        print(f"指定目錄臨時文件: {specific_temp.name}")
    
    # 手動清理
    os.unlink(specific_temp.name)

demonstrate_advanced_tempfile_modes()

class DataProcessingPipeline:
    """
    使用臨時文件的數(shù)據(jù)處理管道
    """
    
    def __init__(self):
        self.temp_files = []
        self.temp_dirs = []
    
    def process_large_dataset(self, data_generator, process_func, chunk_size=1000):
        """
        處理大型數(shù)據(jù)集
        """
        print("=== 大型數(shù)據(jù)處理管道 ===")
        
        # 創(chuàng)建臨時工作目錄
        work_dir = tempfile.mkdtemp(prefix='data_pipeline_')
        self.temp_dirs.append(work_dir)
        print(f"創(chuàng)建工作目錄: {work_dir}")
        
        processed_chunks = []
        chunk_count = 0
        
        try:
            # 分塊處理數(shù)據(jù)
            current_chunk = []
            for item in data_generator:
                current_chunk.append(item)
                
                if len(current_chunk) >= chunk_size:
                    # 處理當前塊
                    processed_chunk = self._process_chunk(current_chunk, process_func, work_dir, chunk_count)
                    processed_chunks.append(processed_chunk)
                    current_chunk = []
                    chunk_count += 1
            
            # 處理最后一塊
            if current_chunk:
                processed_chunk = self._process_chunk(current_chunk, process_func, work_dir, chunk_count)
                processed_chunks.append(processed_chunk)
            
            # 合并結(jié)果
            final_result = self._merge_results(processed_chunks, work_dir)
            print(f"處理完成: {chunk_count + 1} 個數(shù)據(jù)塊")
            
            return final_result
            
        except Exception as e:
            print(f"數(shù)據(jù)處理錯誤: {e}")
            self.cleanup()
            raise
    
    def _process_chunk(self, chunk_data, process_func, work_dir, chunk_index):
        """
        處理單個數(shù)據(jù)塊
        """
        # 創(chuàng)建臨時文件存儲塊數(shù)據(jù)
        chunk_file = tempfile.NamedTemporaryFile(
            mode='w+',
            dir=work_dir,
            suffix=f'_chunk_{chunk_index}.json',
            delete=False,
            encoding='utf-8'
        )
        self.temp_files.append(chunk_file.name)
        
        try:
            # 處理數(shù)據(jù)
            processed_data = process_func(chunk_data)
            
            # 寫入臨時文件
            import json
            json.dump(processed_data, chunk_file, ensure_ascii=False)
            chunk_file.close()
            
            return chunk_file.name
            
        except Exception as e:
            chunk_file.close()
            os.unlink(chunk_file.name)
            raise
    
    def _merge_results(self, chunk_files, work_dir):
        """
        合并處理結(jié)果
        """
        merged_data = []
        
        for chunk_file in chunk_files:
            try:
                with open(chunk_file, 'r', encoding='utf-8') as f:
                    import json
                    chunk_data = json.load(f)
                    merged_data.extend(chunk_data)
            except Exception as e:
                print(f"合并塊 {chunk_file} 錯誤: {e}")
        
        # 創(chuàng)建最終結(jié)果文件
        result_file = tempfile.NamedTemporaryFile(
            mode='w+',
            dir=work_dir,
            suffix='_final_result.json',
            delete=False,
            encoding='utf-8'
        )
        self.temp_files.append(result_file.name)
        
        import json
        json.dump(merged_data, result_file, ensure_ascii=False, indent=2)
        result_file.close()
        
        return result_file.name
    
    def cleanup(self):
        """清理所有臨時資源"""
        for file_path in self.temp_files:
            try:
                if os.path.exists(file_path):
                    os.unlink(file_path)
            except OSError:
                pass
        
        for dir_path in self.temp_dirs:
            try:
                if os.path.exists(dir_path):
                    shutil.rmtree(dir_path)
            except OSError:
                pass
        
        self.temp_files.clear()
        self.temp_dirs.clear()
        print("所有臨時資源已清理")
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup()

# 使用示例
def demo_data_pipeline():
    """數(shù)據(jù)處理管道演示"""
    
    # 模擬數(shù)據(jù)生成器
    def data_generator():
        for i in range(10000):
            yield {'id': i, 'data': f'item_{i}', 'value': i * 2}
    
    # 模擬處理函數(shù)
    def process_function(chunk):
        return [{'processed': item, 'timestamp': time.time()} for item in chunk]
    
    # 使用管道
    with DataProcessingPipeline() as pipeline:
        result_file = pipeline.process_large_dataset(data_generator(), process_function, chunk_size=500)
        
        # 讀取部分結(jié)果
        with open(result_file, 'r', encoding='utf-8') as f:
            import json
            result_data = json.load(f)
            print(f"處理結(jié)果: {len(result_data)} 條記錄")
            print(f"示例記錄: {result_data[0] if result_data else '無數(shù)據(jù)'}")

demo_data_pipeline()

class TestTempFileManager:
    """
    測試用的臨時文件管理器
    """
    
    def __init__(self):
        self.test_files = []
        self.test_dirs = []
    
    def create_test_config(self, config_data):
        """創(chuàng)建測試配置文件"""
        config_file = tempfile.NamedTemporaryFile(
            mode='w+',
            suffix='.json',
            delete=False,
            encoding='utf-8'
        )
        self.test_files.append(config_file.name)
        
        import json
        json.dump(config_data, config_file, indent=2)
        config_file.close()
        
        return config_file.name
    
    def create_test_database(self, db_data):
        """創(chuàng)建測試數(shù)據(jù)庫文件"""
        db_file = tempfile.NamedTemporaryFile(
            mode='wb',
            suffix='.db',
            delete=False
        )
        self.test_files.append(db_file.name)
        
        # 模擬數(shù)據(jù)庫文件創(chuàng)建
        db_file.write(b'SQLite format 3\x00')  # SQLite文件頭
        # 這里可以添加更多的模擬數(shù)據(jù)庫內(nèi)容
        db_file.close()
        
        return db_file.name
    
    def create_test_directory_structure(self, structure):
        """創(chuàng)建測試目錄結(jié)構(gòu)"""
        base_dir = tempfile.mkdtemp(prefix='test_structure_')
        self.test_dirs.append(base_dir)
        
        def create_structure(current_path, current_structure):
            for name, content in current_structure.items():
                item_path = os.path.join(current_path, name)
                
                if isinstance(content, dict):
                    # 創(chuàng)建子目錄
                    os.makedirs(item_path, exist_ok=True)
                    create_structure(item_path, content)
                else:
                    # 創(chuàng)建文件
                    with open(item_path, 'w', encoding='utf-8') as f:
                        f.write(content if isinstance(content, str) else str(content))
        
        create_structure(base_dir, structure)
        return base_dir
    
    def cleanup(self):
        """清理測試文件"""
        for file_path in self.test_files:
            try:
                os.unlink(file_path)
            except OSError:
                pass
        
        for dir_path in self.test_dirs:
            try:
                shutil.rmtree(dir_path)
            except OSError:
                pass
        
        self.test_files.clear()
        self.test_dirs.clear()

# 使用示例
def demo_test_framework():
    """測試框架演示"""
    test_manager = TestTempFileManager()
    
    try:
        # 創(chuàng)建測試配置
        config_data = {
            'database': {
                'host': 'localhost',
                'port': 5432,
                'name': 'test_db'
            },
            'settings': {
                'debug': True,
                'timeout': 30
            }
        }
        config_file = test_manager.create_test_config(config_data)
        print(f"測試配置文件: {config_file}")
        
        # 創(chuàng)建測試數(shù)據(jù)庫
        db_file = test_manager.create_test_database({'test': 'data'})
        print(f"測試數(shù)據(jù)庫文件: {db_file}")
        
        # 創(chuàng)建目錄結(jié)構(gòu)
        dir_structure = {
            'config': {
                'app.ini': '[main]\nversion=1.0',
                'secrets': {
                    'key.txt': 'secret-key-12345'
                }
            },
            'data': {
                'input.csv': 'id,name,value\n1,test,100',
                'output': {}
            },
            'logs': {
                'app.log': 'LOG START\n'
            }
        }
        test_dir = test_manager.create_test_directory_structure(dir_structure)
        print(f"測試目錄結(jié)構(gòu): {test_dir}")
        
        # 顯示創(chuàng)建的內(nèi)容
        for root, dirs, files in os.walk(test_dir):
            level = root.replace(test_dir, '').count(os.sep)
            indent = ' ' * 2 * level
            print(f"{indent}{os.path.basename(root)}/")
            sub_indent = ' ' * 2 * (level + 1)
            for file in files:
                print(f"{sub_indent}{file}")
        
    finally:
        test_manager.cleanup()
        print("測試資源已清理")

demo_test_framework()

class OptimizedTempFileSystem:
    """
    優(yōu)化的臨時文件系統(tǒng)
    """
    
    def __init__(self, max_memory_size=100 * 1024 * 1024):  # 100MB
        self.max_memory_size = max_memory_size
        self.memory_buffer = bytearray()
        self.temp_files = []
        self.total_size = 0
    
    def write_data(self, data):
        """
        智能寫入數(shù)據(jù)，根據(jù)大小選擇內(nèi)存或磁盤存儲
        """
        data_size = len(data) if isinstance(data, bytes) else len(data.encode('utf-8'))
        
        if self.total_size + data_size <= self.max_memory_size:
            # 使用內(nèi)存存儲
            if isinstance(data, bytes):
                self.memory_buffer.extend(data)
            else:
                self.memory_buffer.extend(data.encode('utf-8'))
            self.total_size += data_size
            return None  # 返回None表示數(shù)據(jù)在內(nèi)存中
        else:
            # 使用臨時文件存儲
            temp_file = self._create_temp_file()
            if isinstance(data, bytes):
                temp_file.write(data)
            else:
                temp_file.write(data.encode('utf-8'))
            self.total_size += data_size
            return temp_file.name
    
    def _create_temp_file(self):
        """創(chuàng)建臨時文件"""
        temp_file = tempfile.NamedTemporaryFile(mode='wb', delete=False)
        self.temp_files.append(temp_file.name)
        return temp_file
    
    def read_all_data(self):
        """讀取所有數(shù)據(jù)"""
        # 讀取內(nèi)存數(shù)據(jù)
        all_data = bytes(self.memory_buffer)
        
        # 讀取文件數(shù)據(jù)
        for file_path in self.temp_files:
            try:
                with open(file_path, 'rb') as f:
                    file_data = f.read()
                    all_data += file_data
            except OSError:
                continue
        
        return all_data
    
    def clear(self):
        """清理所有資源"""
        self.memory_buffer.clear()
        self.total_size = 0
        
        for file_path in self.temp_files:
            try:
                os.unlink(file_path)
            except OSError:
                pass
        
        self.temp_files.clear()
    
    def get_stats(self):
        """獲取統(tǒng)計信息"""
        return {
            'total_size': self.total_size,
            'memory_usage': len(self.memory_buffer),
            'file_count': len(self.temp_files),
            'using_disk': len(self.temp_files) > 0
        }

# 使用示例
def demo_optimized_storage():
    """優(yōu)化存儲演示"""
    print("=== 優(yōu)化臨時存儲系統(tǒng) ===")
    
    storage = OptimizedTempFileSystem(max_memory_size=50)  # 小尺寸用于演示
    
    test_data = [
        "小數(shù)據(jù)塊1",
        "小數(shù)據(jù)塊2",
        "這個數(shù)據(jù)塊比較大，應(yīng)該會觸發(fā)文件存儲" * 10,
        "另一個小數(shù)據(jù)塊"
    ]
    
    for i, data in enumerate(test_data):
        storage_location = storage.write_data(data)
        stats = storage.get_stats()
        
        if storage_location:
            print(f"數(shù)據(jù) {i}: 存儲在文件 {storage_location}")
        else:
            print(f"數(shù)據(jù) {i}: 存儲在內(nèi)存中")
        
        print(f"  統(tǒng)計: 總大小={stats['total_size']}, 文件數(shù)={stats['file_count']}")
    
    # 讀取所有數(shù)據(jù)
    all_data = storage.read_all_data()
    print(f"\n總共存儲數(shù)據(jù): {len(all_data)} 字節(jié)")
    
    storage.clear()
    print("存儲系統(tǒng)已清理")

demo_optimized_storage()

def cross_platform_tempfile_considerations():
    """
    跨平臺臨時文件處理注意事項
    """
    print("=== 跨平臺兼容性考慮 ===")
    
    considerations = {
        'Windows': [
            '文件路徑使用反斜杠，需要特殊處理',
            '文件鎖定機制不同',
            '權(quán)限系統(tǒng)基于ACL而非Unix權(quán)限',
            '臨時目錄通常為C:\\Users\\Username\\AppData\\Local\\Temp'
        ],
        'Linux/macOS': [
            '文件路徑使用正斜杠',
            '基于Unix權(quán)限系統(tǒng)',
            '支持符號鏈接和硬鏈接',
            '臨時目錄通常為/tmp或/var/tmp'
        ],
        '通用最佳實踐': [
            '始終使用os.path.join()構(gòu)建路徑',
            '使用tempfile.gettempdir()獲取系統(tǒng)臨時目錄',
            '處理文件權(quán)限差異',
            '考慮文件名長度限制',
            '處理路徑字符編碼差異'
        ]
    }
    
    for platform, notes in considerations.items():
        print(f"\n{platform}:")
        for note in notes:
            print(f"  ? {note}")
    
    # 演示跨平臺臨時目錄獲取
    print(f"\n當前系統(tǒng)臨時目錄: {tempfile.gettempdir()}")
    
    # 演示跨平臺路徑處理
    test_paths = [
        ('config', 'subdir', 'file.txt'),
        ('data', '2023', '12', '31', 'log.json')
    ]
    
    for path_parts in test_paths:
        constructed_path = os.path.join(*path_parts)
        print(f"構(gòu)建的路徑: {constructed_path}")

cross_platform_tempfile_considerations()

class SecureTempFileServer:
    """
    安全臨時文件服務(wù)器
    """
    
    def __init__(self, max_file_age=3600):  # 1小時
        self.base_dir = tempfile.mkdtemp(prefix='secure_server_')
        self.max_file_age = max_file_age
        self.file_registry = {}
        print(f"安全服務(wù)器啟動在: {self.base_dir}")
    
    def upload_file(self, file_data, filename=None, metadata=None):
        """
        上傳文件到安全臨時存儲
        """
        # 生成安全文件名
        if filename is None:
            import uuid
            filename = f"file_{uuid.uuid4().hex}"
        
        # 創(chuàng)建安全文件路徑
        safe_filename = self._sanitize_filename(filename)
        file_path = os.path.join(self.base_dir, safe_filename)
        
        try:
            # 寫入文件 with secure permissions
            with open(file_path, 'wb') as f:
                f.write(file_data if isinstance(file_data, bytes) else file_data.encode('utf-8'))
            
            # 設(shè)置安全權(quán)限
            os.chmod(file_path, 0o600)
            
            # 記錄文件信息
            file_info = {
                'path': file_path,
                'filename': safe_filename,
                'upload_time': time.time(),
                'size': len(file_data),
                'metadata': metadata or {},
                'access_count': 0
            }
            
            self.file_registry[safe_filename] = file_info
            print(f"文件上傳成功: {safe_filename} ({len(file_data)} 字節(jié))")
            
            return safe_filename
            
        except Exception as e:
            print(f"文件上傳失敗: {e}")
            # 清理部分上傳的文件
            if os.path.exists(file_path):
                os.unlink(file_path)
            raise
    
    def download_file(self, file_id):
        """
        下載文件
        """
        if file_id not in self.file_registry:
            raise ValueError("文件不存在")
        
        file_info = self.file_registry[file_id]
        file_info['access_count'] += 1
        file_info['last_access'] = time.time()
        
        try:
            with open(file_info['path'], 'rb') as f:
                content = f.read()
            
            print(f"文件下載: {file_id} (第{file_info['access_count']}次訪問)")
            return content
            
        except Exception as e:
            print(f"文件下載失敗: {e}")
            raise
    
    def cleanup_old_files(self):
        """
        清理過期文件
        """
        current_time = time.time()
        files_to_remove = []
        
        for file_id, file_info in self.file_registry.items():
            file_age = current_time - file_info['upload_time']
            if file_age > self.max_file_age:
                files_to_remove.append(file_id)
        
        for file_id in files_to_remove:
            self._remove_file(file_id)
            print(f"清理過期文件: {file_id}")
        
        return len(files_to_remove)
    
    def _remove_file(self, file_id):
        """移除單個文件"""
        if file_id in self.file_registry:
            file_info = self.file_registry[file_id]
            try:
                if os.path.exists(file_info['path']):
                    os.unlink(file_info['path'])
            except OSError:
                pass
            finally:
                del self.file_registry[file_id]
    
    def _sanitize_filename(self, filename):
        """消毒文件名"""
        import re
        # 移除危險字符
        safe_name = re.sub(r'[^\w\.-]', '_', filename)
        # 限制長度
        if len(safe_name) > 100:
            safe_name = safe_name[:100]
        return safe_name
    
    def shutdown(self):
        """關(guān)閉服務(wù)器并清理所有文件"""
        print("關(guān)閉安全服務(wù)器...")
        
        # 移除所有文件
        for file_id in list(self.file_registry.keys()):
            self._remove_file(file_id)
        
        # 移除基礎(chǔ)目錄
        try:
            shutil.rmtree(self.base_dir)
        except OSError:
            pass
        
        print("服務(wù)器已關(guān)閉，所有資源已清理")

# 使用示例
def demo_secure_server():
    """安全服務(wù)器演示"""
    server = SecureTempFileServer(max_file_age=300)  # 5分鐘用于演示
    
    try:
        # 上傳文件
        test_files = [
            b'binary file content',
            'text file content with special chars ?áéíóú',
            'x' * 1000  # 大文件
        ]
        
        file_ids = []
        for i, content in enumerate(test_files):
            file_id = server.upload_file(content, f'test_file_{i}.dat')
            file_ids.append(file_id)
        
        # 下載文件
        for file_id in file_ids:
            content = server.download_file(file_id)
            print(f"下載 {file_id}: {len(content)} 字節(jié)")
        
        # 等待并清理過期文件
        print("等待文件過期...")
        time.sleep(2)  # 實際應(yīng)用中這里會是更長的時間
        
        cleaned = server.cleanup_old_files()
        print(f"清理了 {cleaned} 個過期文件")
        
    finally:
        server.shutdown()

demo_secure_server()