Python os模块系统交互全攻略:掌握操作系统接口,构建企业级自动化平台!

import os# 实战案例1：安全创建多层目录结构defsetup_project_structure(project_name):"""为项目创建标准目录结构，自动处理已存在的目录"""# 定义项目标准目录结构    directories = [f"{project_name}/src/main/python",f"{project_name}/src/test/python",f"{project_name}/docs/api",f"{project_name}/docs/tutorials",f"{project_name}/data/raw",f"{project_name}/data/processed",f"{project_name}/logs/debug",f"{project_name}/logs/production",f"{project_name}/config/dev",f"{project_name}/config/prod"    ]    created_dirs = []for directory in directories:try:# exist_ok=True参数是关键：如果目录已存在，不会抛出异常            os.makedirs(directory, exist_ok=True)            print(f"✓ 目录已创建/已存在: {directory}")            created_dirs.append(directory)except PermissionError as e:            print(f"✗ 权限不足，无法创建目录 {directory}: {e}")except OSError as e:            print(f"✗ 创建目录失败 {directory}: {e}")return created_dirs# 执行目录创建if __name__ == "__main__":    project_dirs = setup_project_structure("nexus_platform")    print(f"\n共处理{len(project_dirs)}个目录，项目结构已就绪！")# 获取当前工作目录，验证创建结果    cwd = os.getcwd()    print(f"当前工作目录: {cwd}")

import osimport timefrom typing import List, TupleclassFileSystemScanner:"""高性能文件系统扫描器，支持多种统计和分析功能"""def__init__(self, root_path: str):if not os.path.exists(root_path):raise FileNotFoundError(f"路径不存在: {root_path}")        self.root_path = os.path.abspath(root_path)defscan_with_stats(self) -> Tuple[List[str], dict]:"""扫描目录并返回详细统计信息"""        total_files = 0        total_dirs = 0        total_size = 0        file_extensions = {}        file_list = []        start_time = time.time()# os.walk()返回生成器，避免一次性加载所有路径到内存for root, dirs, files in os.walk(self.root_path):            total_dirs += len(dirs)            total_files += len(files)for file in files:                file_path = os.path.join(root, file)                file_list.append(file_path)# 统计文件大小try:                    file_size = os.path.getsize(file_path)                    total_size += file_sizeexcept OSError:continue# 跳过无法访问的文件# 统计文件扩展名分布                _, ext = os.path.splitext(file)                ext = ext.lower() if ext else"无扩展名"                file_extensions[ext] = file_extensions.get(ext, 0) + 1        elapsed_time = time.time() - start_time        stats = {"扫描根目录": self.root_path,"总耗时(秒)": round(elapsed_time, 2),"目录总数": total_dirs,"文件总数": total_files,"总大小(字节)": total_size,"总大小(GB)": round(total_size / (1024**3), 2),"文件类型分布": dict(sorted(file_extensions.items(), key=lambda x: x[1], reverse=True)),"平均扫描速度(文件/秒)": round(total_files / elapsed_time, 2) if elapsed_time > 0else0        }return file_list, statsdeffind_files_by_pattern(self, pattern_func) -> List[str]:"""使用自定义模式函数查找文件"""        matches = []for root, _, files in os.walk(self.root_path):for file in files:if pattern_func(file):                    matches.append(os.path.join(root, file))return matches# 使用示例if __name__ == "__main__":# 扫描当前目录    scanner = FileSystemScanner(".")    files, stats = scanner.scan_with_stats()    print("=== 文件系统扫描报告 ===")for key, value in stats.items():        print(f"{key}: {value}")# 查找所有Python文件    python_files = scanner.find_files_by_pattern(lambda f: f.endswith(".py"))    print(f"\n找到 {len(python_files)} 个Python文件")# 查找大文件（>10MB）    large_files = []for root, _, files in os.walk("."):for file in files:            file_path = os.path.join(root, file)try:if os.path.getsize(file_path) > 10 * 1024 * 1024:  # 10MB                    large_files.append(file_path)except OSError:continue    print(f"\n找到 {len(large_files)} 个大于10MB的文件")

import osimport jsonfrom typing import Dict, OptionalclassEnvironmentManager:"""专业级环境变量管理器，支持验证、类型转换和安全操作"""def__init__(self, env_prefix: str = "APP_"):        self.env_prefix = env_prefixdefget_required(self, key: str, env_key: Optional[str] = None) -> str:"""获取必需的环境变量，不存在时抛出明确异常"""        env_key = env_key orf"{self.env_prefix}{key.upper()}"        value = os.getenv(env_key)if value isNone:raise EnvironmentError(f"必需的环境变量未设置: {env_key}\n"f"请通过以下方式设置：\n"f"  export {env_key}=value\n"f"或在.env文件中添加：\n"f"  {env_key}=value"            )return valuedefget_with_default(self, key: str, default: str, env_key: Optional[str] = None) -> str:"""获取环境变量，不存在时返回默认值"""        env_key = env_key orf"{self.env_prefix}{key.upper()}"return os.getenv(env_key, default)defget_boolean(self, key: str, default: bool = False, env_key: Optional[str] = None) -> bool:"""将环境变量解析为布尔值"""        env_key = env_key orf"{self.env_prefix}{key.upper()}"        value = os.getenv(env_key, "").lower()if value in ("true", "yes", "1", "on"):returnTrueelif value in ("false", "no", "0", "off"):returnFalseelse:return defaultdefget_json(self, key: str, default: Dict = None, env_key: Optional[str] = None) -> Dict:"""解析JSON格式的环境变量"""        env_key = env_key orf"{self.env_prefix}{key.upper()}"        value = os.getenv(env_key)if value isNone:return default or {}try:return json.loads(value)except json.JSONDecodeError as e:raise ValueError(f"环境变量 {env_key} 包含无效的JSON: {e}")defset_temp(self, key: str, value: str) -> None:"""临时设置环境变量（仅当前进程有效）"""        os.environ[key] = valuedefvalidate_environment(self) -> Dict[str, bool]:"""验证所有必需的环境变量是否已设置"""        validation_results = {}# 常见的必需环境变量检查        common_checks = [            ("DATABASE_URL", "数据库连接字符串"),            ("REDIS_URL", "Redis连接字符串"),            ("SECRET_KEY", "应用密钥"),            ("DEBUG_MODE", "调试模式"),        ]for key, description in common_checks:            full_key = f"{self.env_prefix}{key}"            exists = full_key in os.environ            validation_results[full_key] = existsifnot exists:                print(f"⚠️  警告: {description}未设置 ({full_key})")return validation_results# 实战使用示例if __name__ == "__main__":# 初始化环境管理器    env_mgr = EnvironmentManager(env_prefix="MYAPP_")# 设置一些测试环境变量    os.environ["MYAPP_DATABASE_URL"] = "postgresql://user:pass@localhost/db"    os.environ["MYAPP_DEBUG"] = "true"    os.environ["MYAPP_CONFIG"] = '{"timeout": 30, "retries": 3}'# 获取不同类型的环境变量    db_url = env_mgr.get_required("database_url")    debug_mode = env_mgr.get_boolean("debug")    app_config = env_mgr.get_json("config")    print(f"数据库URL: {db_url}")    print(f"调试模式: {debug_mode}")    print(f"应用配置: {app_config}")# 验证环境    validation = env_mgr.validate_environment()    print(f"\n环境验证结果: {validation}")

import osimport subprocessimport sysfrom typing import Tuple, OptionalclassSystemCommandExecutor:"""安全执行系统命令的工具类，支持超时控制、输出捕获和错误处理"""    @staticmethoddefrun_safe(command: str,                  timeout: Optional[int] = 30,                 cwd: Optional[str] = None,                 env: Optional[Dict] = None) -> Tuple[bool, str, str]:"""        安全执行系统命令        参数:            command: 要执行的命令字符串            timeout: 超时时间（秒），None表示无超时            cwd: 工作目录，None表示当前目录            env: 环境变量字典，None表示继承当前环境        返回:            (成功与否, 标准输出, 标准错误)        """# 确保工作目录存在if cwd andnot os.path.exists(cwd):raise FileNotFoundError(f"工作目录不存在: {cwd}")try:# 使用subprocess.run代替os.system，更安全、功能更全            result = subprocess.run(                command,                shell=True,                capture_output=True,                text=True,                timeout=timeout,                cwd=cwd,                env=env or os.environ.copy()            )            success = result.returncode == 0            stdout = result.stdout.strip()            stderr = result.stderr.strip()return success, stdout, stderrexcept subprocess.TimeoutExpired:returnFalse, "", f"命令执行超时（{timeout}秒）"except FileNotFoundError:returnFalse, "", "命令或程序未找到"except PermissionError:returnFalse, "", "权限不足，无法执行命令"except Exception as e:returnFalse, "", f"执行命令时发生未知错误: {e}"    @staticmethoddefget_system_info() -> Dict:"""获取详细的系统信息"""        info = {}# 获取操作系统类型        info["os_name"] = os.name  # 'posix' 或 'nt'# 获取CPU核心数        info["cpu_cores"] = os.cpu_count()# Python 3.13+新增：获取进程可用的CPU核心数try:            info["process_cpu_cores"] = os.process_cpu_count()except AttributeError:            info["process_cpu_cores"] = "需要Python 3.13+"# 获取路径分隔符        info["path_separator"] = os.sep# 获取换行符（跨平台重要！）        info["line_separator"] = repr(os.linesep)# 执行特定系统命令获取更多信息if os.name == 'posix':  # Linux/macOS            success, stdout, _ = SystemCommandExecutor.run_safe("uname -a")if success:                info["uname"] = stdout            success, stdout, _ = SystemCommandExecutor.run_safe("lsb_release -a", timeout=5)if success:                info["distribution"] = stdoutelif os.name == 'nt':  # Windows            success, stdout, _ = SystemCommandExecutor.run_safe("systeminfo | findstr /B /C:\"OS 名称\" /C:\"OS 版本\"", timeout=10)if success:                info["windows_info"] = stdoutreturn info# 实战案例：批量处理文件if __name__ == "__main__":    executor = SystemCommandExecutor()    print("=== 系统信息报告 ===")    sys_info = executor.get_system_info()for key, value in sys_info.items():        print(f"{key}: {value}")    print("\n=== 安全执行命令示例 ===")# 示例1：查看当前目录文件（跨平台）if os.name == 'posix':        command = "ls -la"else:        command = "dir"    success, stdout, stderr = executor.run_safe(command, timeout=5)if success:        print("当前目录文件列表：")        print(stdout[:500] + "..."if len(stdout) > 500else stdout)else:        print(f"命令执行失败: {stderr}")# 示例2：创建备份目录    backup_dir = "./backup_" + str(int(time.time()))    success, stdout, stderr = executor.run_safe(f"mkdir {backup_dir}")if success:        print(f"\n备份目录创建成功: {backup_dir}")else:        print(f"创建备份目录失败: {stderr}")

import osimport sysfrom pathlib import PathclassPathSanitizer:"""跨平台路径处理和安全检查工具"""    @staticmethoddefsafe_join(*parts) -> str:"""安全拼接路径，自动处理跨平台分隔符"""# 使用os.path.join是最佳实践return os.path.join(*parts)    @staticmethoddefnormalize_path(path: str) -> str:"""规范化路径，处理'..'、'.'和多余分隔符"""return os.path.normpath(path)    @staticmethoddefget_absolute_path(path: str) -> str:"""获取绝对路径，解析符号链接"""return os.path.abspath(os.path.expanduser(path))    @staticmethoddefis_safe_path(base_path: str, target_path: str) -> bool:"""检查目标路径是否在基准路径内（防止路径遍历攻击）"""        base = os.path.abspath(base_path)        target = os.path.abspath(target_path)# 确保目标路径以基准路径开头return target.startswith(base + os.sep) or target == base    @staticmethoddefget_file_info(filepath: str) -> Dict:"""获取文件的详细信息"""ifnot os.path.exists(filepath):raise FileNotFoundError(f"文件不存在: {filepath}")        stat_info = os.stat(filepath)return {"路径": filepath,"文件名": os.path.basename(filepath),"目录": os.path.dirname(filepath),"扩展名": os.path.splitext(filepath)[1],"大小(字节)": stat_info.st_size,"创建时间": stat_info.st_ctime,"修改时间": stat_info.st_mtime,"访问时间": stat_info.st_atime,"权限": oct(stat_info.st_mode)[-3:],"是文件": os.path.isfile(filepath),"是目录": os.path.isdir(filepath),"是链接": os.path.islink(filepath),        }    @staticmethoddeffind_files_recursive(root_dir: str,                            pattern: str = None,                           max_depth: int = 10) -> List[str]:"""递归查找文件，支持通配符和深度限制"""        matches = []for depth, (root, dirs, files) in enumerate(os.walk(root_dir)):if depth > max_depth:breakfor file in files:if pattern:import fnmatchifnot fnmatch.fnmatch(file, pattern):continue                matches.append(os.path.join(root, file))return matches# 演示跨平台路径处理的正确方式if __name__ == "__main__":    sanitizer = PathSanitizer()    print("=== 跨平台路径处理演示 ===")# 正确做法：使用os.path.join    path1 = sanitizer.safe_join("user", "data", "config.ini")    print(f"拼接路径: {path1}")# 错误做法：手动拼接（不推荐）# path_wrong = "user" + "/" + "data" + "/" + "config.ini"  # 在Windows上可能失效# 获取文件信息    current_file = __file__if os.path.exists(current_file):        file_info = sanitizer.get_file_info(current_file)        print(f"\n当前文件信息:")for key, value in file_info.items():            print(f"  {key}: {value}")# 安全检查示例    base_dir = "/home/user/projects"    safe_path = "/home/user/projects/data/file.txt"    unsafe_path = "/home/user/projects/../etc/passwd"    print(f"\n路径安全检查:")    print(f"{safe_path} 是否安全: {sanitizer.is_safe_path(base_dir, safe_path)}")    print(f"{unsafe_path} 是否安全: {sanitizer.is_safe_path(base_dir, unsafe_path)}")# 使用pathlib的现代方式（Python 3.4+推荐）    print(f"\n=== 现代pathlib方式 ===")    p = Path(__file__)    print(f"文件名: {p.name}")    print(f"父目录: {p.parent}")    print(f"扩展名: {p.suffix}")    print(f"是否存在: {p.exists()}")

import osimport psutil  # 需要安装：pip install psutilimport platformfrom typing import Dict, TupleclassSystemMonitor:"""系统资源监控器，结合os模块和psutil提供完整信息"""    @staticmethoddefget_basic_info() -> Dict:"""使用os模块获取基本系统信息"""        info = {"操作系统类型": os.name,"CPU逻辑核心数": os.cpu_count(),"路径分隔符": os.sep,"换行符": repr(os.linesep),        }# 平台特定信息if hasattr(os, 'uname'):try:                uname = os.uname()                info["系统名称"] = uname.sysname                info["节点名称"] = uname.nodename                info["内核版本"] = uname.release                info["硬件架构"] = uname.machineexcept AttributeError:passreturn info    @staticmethoddefget_detailed_info() -> Dict:"""使用psutil获取详细系统信息（需要安装psutil）"""        info = {}try:# CPU信息            cpu_percent = psutil.cpu_percent(interval=0.5)            cpu_count_logical = psutil.cpu_count()            cpu_count_physical = psutil.cpu_count(logical=False)            info["CPU使用率(%)"] = cpu_percent            info["CPU逻辑核心数"] = cpu_count_logical            info["CPU物理核心数"] = cpu_count_physical            info["CPU频率(MHz)"] = psutil.cpu_freq().current if psutil.cpu_freq() else"N/A"# 内存信息            memory = psutil.virtual_memory()            info["内存总量(GB)"] = round(memory.total / (1024**3), 2)            info["内存使用(GB)"] = round(memory.used / (1024**3), 2)            info["内存使用率(%)"] = memory.percent            info["可用内存(GB)"] = round(memory.available / (1024**3), 2)# 磁盘信息            disk = psutil.disk_usage('/')            info["磁盘总量(GB)"] = round(disk.total / (1024**3), 2)            info["磁盘使用(GB)"] = round(disk.used / (1024**3), 2)            info["磁盘使用率(%)"] = disk.percent# 网络信息            net_io = psutil.net_io_counters()            info["网络发送(MB)"] = round(net_io.bytes_sent / (1024**2), 2)            info["网络接收(MB)"] = round(net_io.bytes_recv / (1024**2), 2)# 进程信息            info["进程总数"] = len(psutil.pids())except ImportError:            info["错误"] = "需要安装psutil库: pip install psutil"return info    @staticmethoddefcheck_system_health() -> Tuple[bool, Dict]:"""检查系统健康状况"""        health_status = {"CPU使用率": "正常","内存使用率": "正常", "磁盘空间": "正常","系统负载": "正常"        }        all_healthy = Truetry:# 检查CPU使用率            cpu_percent = psutil.cpu_percent(interval=0.5)if cpu_percent > 90:                health_status["CPU使用率"] = f"警告: {cpu_percent}%"                all_healthy = False# 检查内存使用率            memory = psutil.virtual_memory()if memory.percent > 90:                health_status["内存使用率"] = f"警告: {memory.percent}%"                all_healthy = False# 检查磁盘空间            disk = psutil.disk_usage('/')if disk.percent > 90:                health_status["磁盘空间"] = f"警告: {disk.percent}%"                all_healthy = False# 系统负载（仅Linux）if hasattr(os, 'getloadavg'):                load1, load5, load15 = os.getloadavg()                cpu_count = os.cpu_count() or1if load1 > cpu_count * 2:  # 负载超过CPU核心数的2倍                    health_status["系统负载"] = f"警告: {load1:.2f}"                    all_healthy = Falseexcept Exception as e:            health_status["错误"] = str(e)            all_healthy = Falsereturn all_healthy, health_status# 实战：系统监控仪表板if __name__ == "__main__":    monitor = SystemMonitor()    print("=" * 60)    print("系统监控仪表板")    print("=" * 60)# 获取基本信息    basic_info = monitor.get_basic_info()    print("\n[基本系统信息]")for key, value in basic_info.items():        print(f"{key:20}: {value}")# 获取详细信息    detailed_info = monitor.get_detailed_info()    print("\n[详细系统信息]")for key, value in detailed_info.items():        print(f"{key:20}: {value}")# 系统健康检查    healthy, health_status = monitor.check_system_health()    print(f"\n[系统健康状态: {'✅ 健康'if healthy else'⚠️  警告'}]")for key, value in health_status.items():        print(f"{key:20}: {value}")# 性能优化建议    print("\n[性能优化建议]")if'CPU逻辑核心数'in basic_info:        cpu_cores = basic_info['CPU逻辑核心数']        print(f"1. 建议配置的线程数: {cpu_cores * 2} (基于{cpu_cores}个CPU核心)")if'内存总量(GB)'in detailed_info:        memory_gb = detailed_info['内存总量(GB)']if memory_gb < 8:            print("2. ⚠️  内存不足: 建议升级到至少8GB内存")elif memory_gb < 16:            print("2. 内存配置: 可满足大多数应用需求")else:            print("2. 内存配置: 充裕，适合高并发应用")

import osimport sysimport platformclassCrossPlatformFramework:"""跨平台兼容性框架，提供统一的系统接口"""def__init__(self):        self.os_type = os.name  # 'posix' 或 'nt'        self.platform_info = platform.system()        self.architecture = platform.machine()defget_home_directory(self) -> str:"""获取用户主目录（跨平台）"""if self.os_type == 'posix':return os.path.expanduser("~")elif self.os_type == 'nt':return os.environ.get("USERPROFILE", os.environ.get("HOMEDRIVE", "") + os.environ.get("HOMEPATH", ""))else:return os.path.expanduser("~")defget_temp_directory(self) -> str:"""获取临时目录（跨平台）"""        temp_dir = os.environ.get("TEMP") or os.environ.get("TMP") or"/tmp"ifnot os.path.exists(temp_dir):            os.makedirs(temp_dir, exist_ok=True)return temp_dirdefexecute_command(self, command_template: str, **kwargs) -> Tuple[bool, str]:"""执行跨平台命令（自动适配）"""# 根据平台调整命令if self.os_type == 'nt':# Windows命令适配if'ls'in command_template:                command = command_template.replace('ls', 'dir')elif'grep'in command_template:                command = command_template.replace('grep', 'findstr')else:                command = command_templateelse:            command = command_template# 填充模板参数if kwargs:            command = command.format(**kwargs)# 安全执行try:import subprocess            result = subprocess.run(                command,                shell=True,                capture_output=True,                text=True,                timeout=30            )return result.returncode == 0, result.stdoutexcept Exception as e:returnFalse, str(e)defnormalize_file_path(self, path: str) -> str:"""规范化文件路径（解决大小写、分隔符等问题）"""# 转换为绝对路径        abs_path = os.path.abspath(os.path.expanduser(path))# 规范化路径分隔符        normalized = os.path.normpath(abs_path)# Windows上的大小写处理if self.os_type == 'nt':try:# 获取实际大小写（如果文件存在）import ctypesfrom ctypes import wintypes                kernel32 = ctypes.windll.kernel32                GetLongPathNameW = kernel32.GetLongPathNameW                buf = ctypes.create_unicode_buffer(260)if GetLongPathNameW(normalized, buf, 260):                    normalized = buf.valueexcept:pass# 失败时使用原路径return normalizeddefcreate_platform_specific_shortcut(self, target_path: str, shortcut_path: str):"""创建平台特定的快捷方式/符号链接"""if self.os_type == 'posix':# Unix符号链接if os.path.exists(shortcut_path):                os.remove(shortcut_path)            os.symlink(target_path, shortcut_path)elif self.os_type == 'nt':# Windows快捷方式 (.lnk)try:import winshellfrom win32com.client import Dispatch                shell = Dispatch('WScript.Shell')                shortcut = shell.CreateShortCut(shortcut_path)                shortcut.Targetpath = target_path                shortcut.save()except ImportError:                print("Windows快捷方式需要pywin32库: pip install pywin32")

import osimport sysimport shutilimport tempfileimport hashlibfrom typing import List, DictclassDeploymentAutomation:"""企业级自动化部署系统"""def__init__(self, project_root: str):        self.project_root = os.path.abspath(project_root)        self.backup_dir = os.path.join(self.project_root, ".deploy_backups")defcreate_backup(self) -> str:"""创建部署前的完整备份"""        timestamp = time.strftime("%Y%m%d_%H%M%S")        backup_path = os.path.join(self.backup_dir, f"backup_{timestamp}")        os.makedirs(self.backup_dir, exist_ok=True)# 复制整个项目目录        shutil.copytree(self.project_root, backup_path,                        ignore=shutil.ignore_patterns("*.pyc", "__pycache__", ".git"))        print(f"✅ 备份创建完成: {backup_path}")return backup_pathdefdeploy_files(self, source_dir: str, file_pattern: str = "**/*") -> Dict:"""部署文件到目标目录"""        results = {"total_files": 0,"success": 0,"failed": 0,"skipped": 0,"errors": []        }        source_path = os.path.abspath(source_dir)for root, dirs, files in os.walk(source_path):for file in files:                source_file = os.path.join(root, file)                rel_path = os.path.relpath(source_file, source_path)                target_file = os.path.join(self.project_root, rel_path)                results["total_files"] += 1try:# 确保目标目录存在                    os.makedirs(os.path.dirname(target_file), exist_ok=True)# 检查是否需要更新（比较MD5）if os.path.exists(target_file):                        source_md5 = self._calculate_md5(source_file)                        target_md5 = self._calculate_md5(target_file)if source_md5 == target_md5:                            results["skipped"] += 1continue# 复制文件                    shutil.copy2(source_file, target_file)                    results["success"] += 1except Exception as e:                    results["failed"] += 1                    results["errors"].append(f"{rel_path}: {str(e)}")return resultsdef_calculate_md5(self, filepath: str) -> str:"""计算文件的MD5哈希值"""        hash_md5 = hashlib.md5()with open(filepath, "rb") as f:for chunk in iter(lambda: f.read(4096), b""):                hash_md5.update(chunk)return hash_md5.hexdigest()defverify_deployment(self) -> bool:"""验证部署结果"""        required_dirs = ["src","config", "logs","data"        ]        required_files = ["requirements.txt","README.md","config/settings.py"        ]        all_passed = True        print("=== 部署验证 ===")# 检查目录for dir_name in required_dirs:            dir_path = os.path.join(self.project_root, dir_name)if os.path.exists(dir_path) and os.path.isdir(dir_path):                print(f"✅ 目录检查通过: {dir_name}")else:                print(f"❌ 目录不存在: {dir_name}")                all_passed = False# 检查文件for file_name in required_files:            file_path = os.path.join(self.project_root, file_name)if os.path.exists(file_path) and os.path.isfile(file_path):                print(f"✅ 文件检查通过: {file_name}")else:                print(f"❌ 文件不存在: {file_name}")                all_passed = Falsereturn all_passed

import osimport timeimport gzipimport shutilfrom datetime import datetime, timedeltafrom typing import ListclassSmartLogManager:"""智能日志管理系统，支持轮转、压缩、归档"""def__init__(self, log_dir: str, max_size_mb: int = 100, keep_days: int = 30):        self.log_dir = os.path.abspath(log_dir)        self.max_size_bytes = max_size_mb * 1024 * 1024        self.keep_days = keep_days        os.makedirs(self.log_dir, exist_ok=True)defrotate_log(self, log_file: str) -> None:"""日志轮转：当文件超过大小时自动轮转"""        full_path = os.path.join(self.log_dir, log_file)ifnot os.path.exists(full_path):return        file_size = os.path.getsize(full_path)if file_size < self.max_size_bytes:return# 生成带时间戳的新文件名        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")        rotated_name = f"{log_file}.{timestamp}"# 重命名原文件        os.rename(full_path, os.path.join(self.log_dir, rotated_name))# 压缩旧日志        self._compress_log(rotated_name)        print(f"✅ 日志轮转完成: {log_file} -> {rotated_name}.gz")def_compress_log(self, filename: str) -> None:"""压缩日志文件"""        source = os.path.join(self.log_dir, filename)        target = f"{source}.gz"with open(source, 'rb') as f_in:with gzip.open(target, 'wb') as f_out:                shutil.copyfileobj(f_in, f_out)# 删除原文件        os.remove(source)defcleanup_old_logs(self) -> List[str]:"""清理过期的日志文件"""        deleted_files = []        cutoff_time = datetime.now() - timedelta(days=self.keep_days)for file in os.listdir(self.log_dir):            file_path = os.path.join(self.log_dir, file)# 只处理压缩的日志文件ifnot file.endswith('.gz'):continue# 获取文件修改时间            mtime = datetime.fromtimestamp(os.path.getmtime(file_path))if mtime < cutoff_time:                os.remove(file_path)                deleted_files.append(file)                print(f"🗑️  删除过期日志: {file}")return deleted_filesdefget_log_statistics(self) -> Dict:"""获取日志统计信息"""        stats = {"total_logs": 0,"total_size_gb": 0,"by_type": {},"oldest_log": None,"newest_log": None        }for file in os.listdir(self.log_dir):            file_path = os.path.join(self.log_dir, file)ifnot os.path.isfile(file_path):continue            file_size = os.path.getsize(file_path)            mtime = datetime.fromtimestamp(os.path.getmtime(file_path))            stats["total_logs"] += 1            stats["total_size_gb"] += file_size# 按类型统计            ext = os.path.splitext(file)[1]            stats["by_type"][ext] = stats["by_type"].get(ext, 0) + 1# 更新最早/最晚时间if stats["oldest_log"] isNoneor mtime < stats["oldest_log"]:                stats["oldest_log"] = mtimeif stats["newest_log"] isNoneor mtime > stats["newest_log"]:                stats["newest_log"] = mtime# 转换为GB        stats["total_size_gb"] = round(stats["total_size_gb"] / (1024**3), 2)return stats

import osimport jsonimport yamlimport configparserfrom typing import Any, Dict, OptionalclassConfigManager:"""跨平台配置文件管理器，支持多种格式"""def__init__(self, config_dir: Optional[str] = None):        self.config_dir = config_dir or self._get_default_config_dir()        os.makedirs(self.config_dir, exist_ok=True)def_get_default_config_dir(self) -> str:"""获取平台特定的配置目录"""if os.name == 'posix':# Linux/macOS: ~/.config/return os.path.expanduser("~/.config/myapp")elif os.name == 'nt':# Windows: %APPDATA%return os.path.join(os.environ.get("APPDATA", ""), "MyApp")else:return"./config"defload_config(self, filename: str, format: str = "auto") -> Dict:"""加载配置文件，自动检测格式"""        filepath = os.path.join(self.config_dir, filename)ifnot os.path.exists(filepath):return {}# 自动检测格式if format == "auto":if filename.endswith('.json'):                format = 'json'elif filename.endswith('.yaml') or filename.endswith('.yml'):                format = 'yaml'elif filename.endswith('.ini') or filename.endswith('.cfg'):                format = 'ini'else:# 默认尝试JSON                format = 'json'try:if format == 'json':with open(filepath, 'r', encoding='utf-8') as f:return json.load(f)elif format == 'yaml':with open(filepath, 'r', encoding='utf-8') as f:return yaml.safe_load(f)elif format == 'ini':                config = configparser.ConfigParser()                config.read(filepath, encoding='utf-8')return {section: dict(config.items(section)) for section in config.sections()}else:raise ValueError(f"不支持的格式: {format}")except Exception as e:            print(f"加载配置文件失败 {filename}: {e}")return {}defsave_config(self, filename: str, data: Dict, format: str = "json") -> bool:"""保存配置文件"""        filepath = os.path.join(self.config_dir, filename)try:if format == 'json':with open(filepath, 'w', encoding='utf-8') as f:                    json.dump(data, f, indent=2, ensure_ascii=False)elif format == 'yaml':with open(filepath, 'w', encoding='utf-8') as f:                    yaml.dump(data, f, default_flow_style=False, allow_unicode=True)elif format == 'ini':                config = configparser.ConfigParser()for section, items in data.items():                    config[section] = itemswith open(filepath, 'w', encoding='utf-8') as f:                    config.write(f)else:raise ValueError(f"不支持的格式: {format}")            print(f"✅ 配置保存成功: {filename}")returnTrueexcept Exception as e:            print(f"保存配置文件失败 {filename}: {e}")returnFalsedefget_platform_specific_config(self) -> Dict:"""获取平台特定的默认配置"""        base_config = {"app": {"name": "MyApplication","version": "1.0.0","debug": False            },"database": {"host": "localhost","port": 5432,"name": "myapp_db"            }        }# 平台特定调整if os.name == 'nt':            base_config["paths"] = {"data_dir": os.path.join(os.environ.get("PROGRAMDATA", ""), "MyApp"),"log_dir": os.path.join(os.environ.get("LOCALAPPDATA", ""), "MyApp", "logs")            }else:            base_config["paths"] = {"data_dir": "/var/lib/myapp","log_dir": "/var/log/myapp"            }return base_config

# 错误做法：直接拼接用户输入defunsafe_execute(command: str):    os.system(f"echo {command}")  # 危险！用户可能输入"; rm -rf /"# 正确做法：使用参数化defsafe_execute(command: str, args: List[str]):import subprocess    subprocess.run([command] + args)  # 安全# 或者使用白名单验证defvalidate_command(command: str) -> bool:    allowed_commands = {"ls", "pwd", "date", "whoami"}return command in allowed_commands

defsanitize_path(user_input: str, base_dir: str) -> Optional[str]:"""防止路径遍历攻击"""# 规范化路径    requested = os.path.normpath(user_input)# 获取绝对路径    requested_abs = os.path.abspath(requested)    base_abs = os.path.abspath(base_dir)# 确保请求的路径在基准目录内ifnot requested_abs.startswith(base_abs):        print(f"❌ 路径安全检查失败: {requested}")returnNonereturn requested_abs

defrun_with_minimal_privileges():"""以最小必要权限运行"""import pwdimport grp# 获取当前用户    current_uid = os.getuid()# 如果不是root，已经是最小权限if current_uid != 0:return# 降权到普通用户    target_user = "nobody"try:        pw = pwd.getpwnam(target_user)        os.setgid(pw.pw_gid)        os.setuid(pw.pw_uid)        print(f"✅ 已降权到用户: {target_user}")except Exception as e:        print(f"降权失败: {e}")

# 正确做法：使用sys.getfilesystemencoding()encoding = sys.getfilesystemencoding()path = "中文目录".encode(encoding) if encoding else"中文目录"# 或者使用pathlib（自动处理编码）from pathlib import Pathp = Path("中文目录")

for root, dirs, files in os.walk(path):# 移除无权限的目录    dirs[:] = [d for d in dirs if os.access(os.path.join(root, d), os.R_OK)]for file in files:try:            filepath = os.path.join(root, file)# 尝试访问with open(filepath, 'r') as f:passexcept PermissionError:            print(f"跳过无权限文件: {filepath}")continue