#!/usr/bin/env python3 """ 文件上传下载功能模块演示脚本 展示文件处理的核心功能。 """ import os import tempfile from core.file_handler import ( FileUploader, FileDownloader, FileTypeValidator, FileSizeValidator, FilenameSanitizer, FileStorageManager, ) def demo_file_upload(): """演示文件上传""" print("\n" + "="*60) print("演示1: 文件上传") print("="*60) uploader = FileUploader(upload_dir="/tmp/demo_uploads") # 创建测试文件 with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: f.write("测试文件内容") test_file_path = f.name # 上传文件 with open(test_file_path, 'rb') as f: result = uploader.upload(f, filename="test.txt") print(f"✅ 上传结果: {result.success}") print(f"✅ 文件ID: {result.file_id}") print(f"✅ 文件名: {result.filename}") print(f"✅ 文件大小: {result.size} bytes") # 清理 os.unlink(test_file_path) return result.file_id, uploader._storage def demo_file_download(file_id, storage): """演示文件下载""" print("\n" + "="*60) print("演示2: 文件下载") print("="*60) downloader = FileDownloader(storage_manager=storage) result = downloader.download(file_id) print(f"✅ 下载结果: {result.success}") if result.content: print(f"✅ 文件内容: {result.content.decode('utf-8')}") def demo_file_type_validation(): """演示文件类型验证""" print("\n" + "="*60) print("演示3: 文件类型验证") print("="*60) validator = FileTypeValidator(allowed_extensions=['.txt', '.pdf', '.jpg']) test_files = [ ("document.txt", True), ("image.jpg", True), ("script.exe", False), ("virus.bat", False), ] for filename, expected in test_files: is_valid = validator.validate(filename) status = "✅" if is_valid == expected else "❌" print(f"{status} {filename}: {'允许' if is_valid else '拒绝'}") def demo_file_size_validation(): """演示文件大小验证""" print("\n" + "="*60) print("演示4: 文件大小验证") print("="*60) validator = FileSizeValidator(max_size=1024) # 1KB test_sizes = [ (512, True, "512 bytes"), (1024, True, "1KB"), (2048, False, "2KB"), ] for size, expected, desc in test_sizes: is_valid = validator.validate(size) status = "✅" if is_valid == expected else "❌" print(f"{status} {desc}: {'允许' if is_valid else '拒绝'}") def demo_filename_sanitization(): """演示文件名净化""" print("\n" + "="*60) print("演示5: 文件名净化") print("="*60) sanitizer = FilenameSanitizer() test_names = [ "../../../etc/passwd", "file;rm -rf /|.txt", "normal_file.txt", ".txt", ] for name in test_names: safe_name = sanitizer.sanitize(name) print(f"✅ 原始: {name[:40]:<40}") print(f" 净化: {safe_name}") def demo_file_storage(): """演示文件存储管理""" print("\n" + "="*60) print("演示6: 文件存储管理") print("="*60) manager = FileStorageManager(storage_dir="/tmp/demo_storage") # 保存文件 file_id = manager.save( "存储测试内容".encode('utf-8'), filename="stored.txt", metadata={"author": "demo", "tags": ["test"]} ) print(f"✅ 保存文件,ID: {file_id}") # 获取文件 content = manager.get(file_id) print(f"✅ 获取文件内容: {content.decode('utf-8')}") # 获取元数据 metadata = manager.get_metadata(file_id) print(f"✅ 元数据: {metadata}") # 删除文件 deleted = manager.delete(file_id) print(f"✅ 删除文件: {deleted}") def demo_batch_upload(): """演示批量上传""" print("\n" + "="*60) print("演示7: 批量上传") print("="*60) uploader = FileUploader(upload_dir="/tmp/demo_batch") # 创建多个测试文件 files = [] for i in range(3): with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: f.write(f"文件{i}内容") files.append(f.name) # 批量上传 results = uploader.upload_batch(files) print(f"✅ 批量上传: {len(results)}个文件") for i, result in enumerate(results): print(f" 文件{i}: {result.success} (ID: {result.file_id})") # 清理 for f in files: if os.path.exists(f): os.unlink(f) def main(): """主函数""" print("\n" + "="*60) print("文件上传下载功能模块演示") print("="*60) file_id, storage = demo_file_upload() demo_file_download(file_id, storage) demo_file_type_validation() demo_file_size_validation() demo_filename_sanitization() demo_file_storage() demo_batch_upload() print("\n" + "="*60) print("✅ 所有演示完成!") print("="*60) if __name__ == "__main__": main()