Python操作系统常用命令超详细指南

Python操作系统常用命令超详细指南

一、引言与概述

在Python中操作和管理操作系统是一项核心技能，无论是文件处理、进程管理还是系统监控，都离不开对操作系统功能的调用。Python通过os、shutil、subprocess、sys、pathlib等多个内置模块提供了丰富的系统操作功能。

本文将全面介绍Python中常用的操作系统命令，涵盖文件操作、目录管理、进程控制、环境变量、路径处理等各个方面，并提供详细的代码示例和最佳实践建议。

二、文件操作命令

1. 文件创建与读写

python

import os import shutil # 1.1 创建空文件 open('new_file.txt', 'w').close() # 方法1 with open('another_file.txt', 'w') as f: pass # 方法2 # 1.2 写入文件 with open('example.txt', 'w') as f: f.write('Hello, World!\n') f.write('This is a test file.\n') f.writelines(['Line 1\n', 'Line 2\n', 'Line 3\n']) # 1.3 读取文件 with open('example.txt', 'r') as f: content = f.read() # 读取全部内容 print("Full content:", content) with open('example.txt', 'r') as f: lines = f.readlines() # 按行读取为列表 print("Lines:", lines) with open('example.txt', 'r') as f: for line in f: # 逐行读取（内存友好） print("Line:", line.strip()) # 1.4 追加内容 with open('example.txt', 'a') as f: f.write('This is appended content.\n') # 1.5 二进制文件操作 with open('binary_file.bin', 'wb') as f: f.write(b'\x00\x01\x02\x03\x04') with open('binary_file.bin', 'rb') as f: data = f.read() print("Binary data:", data)

2. 文件属性与信息

python

import os import time import stat # 2.1 检查文件是否存在 print("File exists:", os.path.exists('example.txt')) # 2.2 获取文件大小 print("File size:", os.path.getsize('example.txt'), "bytes") # 2.3 获取文件修改时间 mod_time = os.path.getmtime('example.txt') print("Last modified:", time.ctime(mod_time)) # 2.4 获取文件创建时间（Windows） if os.name == 'nt': creation_time = os.path.getctime('example.txt') print("Created:", time.ctime(creation_time)) # 2.5 获取文件权限信息 file_stat = os.stat('example.txt') print("File mode:", oct(file_stat.st_mode)) print("Is readable:", os.access('example.txt', os.R_OK)) print("Is writable:", os.access('example.txt', os.W_OK)) print("Is executable:", os.access('example.txt', os.X_OK)) # 2.6 判断文件类型 print("Is file:", os.path.isfile('example.txt')) print("Is directory:", os.path.isdir('example.txt')) print("Is link:", os.path.islink('example.txt'))

3. 文件复制、移动与删除

python

import os import shutil # 3.1 复制文件 shutil.copy2('source.txt', 'destination.txt') # 保留元数据 shutil.copy('source.txt', 'backup/') # 复制到目录 # 3.2 复制文件并保留所有元数据 shutil.copy2('source.txt', 'dest_with_metadata.txt') # 3.3 复制目录树 shutil.copytree('source_dir', 'destination_dir') # 3.4 移动/重命名文件 shutil.move('old_name.txt', 'new_name.txt') # 重命名 shutil.move('file.txt', 'target_dir/') # 移动 # 3.5 删除文件 os.remove('file_to_delete.txt') # 3.6 安全删除（检查是否存在） if os.path.exists('file.txt'): os.remove('file.txt') else: print("File does not exist") # 3.7 删除目录树 shutil.rmtree('directory_to_remove')

4. 文件查找与匹配

python

import os import fnmatch import glob # 4.1 使用glob查找文件 txt_files = glob.glob('*.txt') print("Text files:", txt_files) all_files = glob.glob('**/*', recursive=True) # 递归查找 print("All files:", all_files) # 4.2 使用fnmatch进行模式匹配 files = os.listdir('.') txt_files = fnmatch.filter(files, '*.txt') print("Text files (fnmatch):", txt_files) # 4.3 递归查找特定扩展名文件 def find_files(extension, directory='.'): matches = [] for root, dirnames, filenames in os.walk(directory): for filename in filenames: if filename.endswith(extension): matches.append(os.path.join(root, filename)) return matches py_files = find_files('.py') print("Python files:", py_files) # 4.4 查找特定大小的文件 def find_files_by_size(min_size, max_size, directory='.'): matches = [] for root, dirnames, filenames in os.walk(directory): for filename in filenames: filepath = os.path.join(root, filename) size = os.path.getsize(filepath) if min_size <= size <= max_size: matches.append((filepath, size)) return matches medium_files = find_files_by_size(1024, 10240) # 1KB-10KB

三、目录操作命令

1. 目录创建与删除

python

import os import shutil # 1.1 创建单个目录 os.mkdir('new_directory') # 1.2 创建多级目录 os.makedirs('path/to/nested/directory', exist_ok=True) # 1.3 删除空目录 os.rmdir('empty_directory') # 1.4 递归删除目录 shutil.rmtree('directory_with_content') # 1.5 安全创建目录 def safe_mkdir(dir_path): if not os.path.exists(dir_path): os.makedirs(dir_path) print(f"Created directory: {dir_path}") else: print(f"Directory already exists: {dir_path}") safe_mkdir('my_directory')

2. 目录遍历与列表

python

import os # 2.1 列出目录内容 print("List dir:", os.listdir('.')) # 2.2 递归遍历目录树 for root, dirs, files in os.walk('.'): print(f"Root: {root}") print(f" Directories: {dirs}") print(f" Files: {files}") print("-" * 40) # 2.3 获取当前工作目录 current_dir = os.getcwd() print("Current directory:", current_dir) # 2.4 更改工作目录 os.chdir('..') print("Changed to:", os.getcwd()) os.chdir(current_dir) # 切换回来 # 2.5 获取绝对路径 abs_path = os.path.abspath('example.txt') print("Absolute path:", abs_path)

3. 路径操作与处理

python

import os from pathlib import Path # 3.1 传统os.path方法 filepath = '/home/user/docs/file.txt' print("Directory:", os.path.dirname(filepath)) print("Filename:", os.path.basename(filepath)) print("Split:", os.path.split(filepath)) print("Splitext:", os.path.splitext(filepath)) # 3.2 路径连接 base_path = '/home/user' full_path = os.path.join(base_path, 'docs', 'file.txt') print("Joined path:", full_path) # 3.3 规范化路径 raw_path = '/home/user/../user/docs/./file.txt' normalized = os.path.normpath(raw_path) print("Normalized:", normalized) # 3.4 使用pathlib（Python 3.4+） path = Path('/home/user/docs/file.txt') print("Path object:", path) print("Parent:", path.parent) print("Name:", path.name) print("Stem:", path.stem) print("Suffix:", path.suffix) # 3.5 相对路径 print("Relative to /home:", path.relative_to('/home')) # 3.6 路径判断 print("Exists:", path.exists()) print("Is absolute:", path.is_absolute()) print("Is file:", path.is_file()) print("Is directory:", path.is_dir())

四、进程管理命令

1. 执行外部命令

python

import subprocess import sys # 1.1 简单执行命令（阻塞） result = subprocess.run(['ls', '-la'], capture_output=True, text=True) print("Return code:", result.returncode) print("Output:", result.stdout) print("Errors:", result.stderr) # 1.2 执行带管道的命令 ps = subprocess.Popen(['ps', 'aux'], stdout=subprocess.PIPE) grep = subprocess.Popen(['grep', 'python'], stdin=ps.stdout, stdout=subprocess.PIPE, text=True) ps.stdout.close() output = grep.communicate()[0] print("Python processes:", output) # 1.3 执行shell命令 result = subprocess.run('echo $HOME && ls -la', shell=True, capture_output=True, text=True) print("Shell output:", result.stdout) # 1.4 超时控制 try: result = subprocess.run(['sleep', '10'], timeout=5, # 5秒超时 capture_output=True) except subprocess.TimeoutExpired: print("Command timed out!") # 1.5 实时输出 process = subprocess.Popen(['ping', '-c', '4', 'google.com'], stdout=subprocess.PIPE, text=True) for line in process.stdout: print(line.strip())

2. 进程信息与控制

python

import os import signal import psutil # 需要安装：pip install psutil # 2.1 获取当前进程ID print("Current PID:", os.getpid()) print("Parent PID:", os.getppid()) # 2.2 获取进程组ID print("Process group:", os.getpgid(0)) # 2.3 创建子进程（fork方式） pid = os.fork() if pid > 0: print(f"Parent process, child PID: {pid}") elif pid == 0: print("Child process") os._exit(0) # 子进程退出 # 2.4 发送信号（需要等待fork完成） import time time.sleep(1) # 2.5 使用psutil获取详细进程信息 for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']): try: if proc.info['cpu_percent'] > 5.0: # CPU使用率>5% print(f"High CPU process: {proc.info}") except (psutil.NoSuchProcess, psutil.AccessDenied): pass # 2.6 终止进程 process = subprocess.Popen(['sleep', '100']) print(f"Started process with PID: {process.pid}") process.terminate() # 发送SIGTERM process.wait() print("Process terminated")

3. 多进程编程

python

import multiprocessing import os import time # 3.1 简单多进程 def worker(num): print(f'Worker {num}: PID {os.getpid()}') return num * 2 if __name__ == '__main__': with multiprocessing.Pool(processes=4) as pool: results = pool.map(worker, range(10)) print("Results:", results) # 3.2 进程间通信 def producer(queue): for i in range(5): queue.put(f'Message {i}') time.sleep(0.1) queue.put(None) # 结束信号 def consumer(queue): while True: item = queue.get() if item is None: break print(f'Consumed: {item}') if __name__ == '__main__': queue = multiprocessing.Queue() p1 = multiprocessing.Process(target=producer, args=(queue,)) p2 = multiprocessing.Process(target=consumer, args=(queue,)) p1.start() p2.start() p1.join() p2.join()

五、系统信息与监控

1. 系统基本信息

python

import os import platform import sys import socket # 1.1 操作系统信息 print("Platform:", platform.platform()) print("System:", platform.system()) print("Release:", platform.release()) print("Version:", platform.version()) print("Architecture:", platform.architecture()) print("Machine:", platform.machine()) print("Processor:", platform.processor()) # 1.2 Python环境信息 print("Python version:", platform.python_version()) print("Python implementation:", platform.python_implementation()) print("Python compiler:", platform.python_compiler()) # 1.3 主机信息 print("Hostname:", socket.gethostname()) print("IP Address:", socket.gethostbyname(socket.gethostname())) # 1.4 登录用户信息 print("Current user:", os.getlogin()) print("User ID:", os.getuid()) print("Group ID:", os.getgid()) print("Effective user ID:", os.geteuid())

2. 磁盘与文件系统

python

import os import shutil import psutil # 需要安装 # 2.1 磁盘使用情况 for partition in psutil.disk_partitions(): print(f"Device: {partition.device}") print(f" Mountpoint: {partition.mountpoint}") print(f" Filesystem: {partition.fstype}") try: usage = psutil.disk_usage(partition.mountpoint) print(f" Total: {usage.total / (1024**3):.2f} GB") print(f" Used: {usage.used / (1024**3):.2f} GB") print(f" Free: {usage.free / (1024**3):.2f} GB") print(f" Usage: {usage.percent}%") except PermissionError: print(" Permission denied") print() # 2.2 获取临时目录 print("Temp dir:", os.environ.get('TEMP', os.environ.get('TMP', '/tmp'))) print("System temp dir:", tempfile.gettempdir()) # 2.3 获取家目录 print("Home directory:", os.path.expanduser('~'))

3. 内存与CPU监控

python

import psutil import time # 3.1 内存信息 memory = psutil.virtual_memory() print(f"Total memory: {memory.total / (1024**3):.2f} GB") print(f"Available: {memory.available / (1024**3):.2f} GB") print(f"Used: {memory.used / (1024**3):.2f} GB") print(f"Usage: {memory.percent}%") swap = psutil.swap_memory() print(f"Swap total: {swap.total / (1024**3):.2f} GB") print(f"Swap used: {swap.used / (1024**3):.2f} GB") # 3.2 CPU信息 print(f"Physical cores: {psutil.cpu_count(logical=False)}") print(f"Logical cores: {psutil.cpu_count(logical=True)}") print(f"CPU frequency: {psutil.cpu_freq().current:.2f} MHz") # 3.3 实时监控 print("Monitoring CPU and memory for 5 seconds...") print("Time\tCPU%\tMemory%") for i in range(5): cpu_percent = psutil.cpu_percent(interval=1) memory_percent = psutil.virtual_memory().percent print(f"{i+1}s\t{cpu_percent:.1f}%\t{memory_percent:.1f}%")

六、环境变量与配置

1. 环境变量操作

python

import os # 1.1 获取所有环境变量 print("All environment variables:") for key, value in os.environ.items(): print(f"{key}: {value}") # 1.2 获取特定环境变量 print("PATH:", os.environ.get('PATH', 'Not set')) print("HOME:", os.environ.get('HOME', 'Not set')) print("PYTHONPATH:", os.environ.get('PYTHONPATH', 'Not set')) # 1.3 设置环境变量（当前进程） os.environ['MY_VARIABLE'] = 'my_value' print("MY_VARIABLE:", os.environ['MY_VARIABLE']) # 1.4 更新环境变量（如PATH） current_path = os.environ.get('PATH', '') new_path = '/usr/local/bin:' + current_path os.environ['PATH'] = new_path # 1.5 删除环境变量 if 'MY_VARIABLE' in os.environ: del os.environ['MY_VARIABLE']

2. 配置文件管理

python

import configparser import json import yaml # 需要安装：pip install pyyaml # 2.1 INI文件处理 config = configparser.ConfigParser() # 读取INI文件 config.read('config.ini') # 获取配置 if 'database' in config: host = config['database']['host'] port = config['database'].getint('port', fallback=3306) print(f"Database: {host}:{port}") # 写入INI文件 config['settings'] = { 'debug': 'true', 'log_level': 'INFO', 'max_connections': '100' } with open('new_config.ini', 'w') as f: config.write(f) # 2.2 JSON配置文件 config_data = { "database": { "host": "localhost", "port": 5432, "name": "mydb" }, "server": { "host": "0.0.0.0", "port": 8000 } } # 写入JSON with open('config.json', 'w') as f: json.dump(config_data, f, indent=2) # 读取JSON with open('config.json', 'r') as f: loaded_config = json.load(f) print("Loaded config:", loaded_config) # 2.3 YAML配置文件（如果安装了pyyaml） try: with open('config.yaml', 'w') as f: yaml.dump(config_data, f, default_flow_style=False) with open('config.yaml', 'r') as f: yaml_config = yaml.safe_load(f) print("YAML config:", yaml_config) except ImportError: print("PyYAML not installed, skipping YAML examples")

七、高级系统操作

1. 文件权限与所有权

python

import os import stat import pwd import grp # 1.1 更改文件权限 os.chmod('myfile.txt', stat.S_IRUSR | stat.S_IWUSR) # 用户读写 os.chmod('script.sh', 0o755) # rwxr-xr-x # 1.2 符号权限表示 permissions = { 'read': stat.S_IRUSR, 'write': stat.S_IWUSR, 'execute': stat.S_IXUSR } # 1.3 获取文件所有者信息 file_stat = os.stat('myfile.txt') print("Owner UID:", file_stat.st_uid) print("Group GID:", file_stat.st_gid) # 1.4 获取用户名和组名 try: owner_name = pwd.getpwuid(file_stat.st_uid).pw_name group_name = grp.getgrgid(file_stat.st_gid).gr_name print(f"Owner: {owner_name}, Group: {group_name}") except (ImportError, KeyError): print("pwd/grp modules may not be available on Windows") # 1.5 更改文件所有者（需要root权限） # os.chown('myfile.txt', uid, gid)

2. 符号链接与硬链接

python

import os # 2.1 创建符号链接 os.symlink('target.txt', 'link.txt') # 2.2 创建硬链接 os.link('original.txt', 'hardlink.txt') # 2.3 读取链接目标 if os.path.islink('link.txt'): target = os.readlink('link.txt') print(f"Link target: {target}") # 2.4 检查链接 print("Is symlink:", os.path.islink('link.txt')) print("Is hardlink:", os.stat('original.txt').st_nlink > 1)

3. 终端与颜色输出

python

import sys import tty import termios import shutil # 3.1 获取终端大小 columns, rows = shutil.get_terminal_size() print(f"Terminal: {columns}x{rows}") # 3.2 颜色输出 class Colors: RED = '\033[91m' GREEN = '\033[92m' YELLOW = '\033[93m' BLUE = '\033[94m' MAGENTA = '\033[95m' CYAN = '\033[96m' END = '\033[0m' BOLD = '\033[1m' print(f"{Colors.RED}Red text{Colors.END}") print(f"{Colors.GREEN}{Colors.BOLD}Bold green text{Colors.END}") # 3.3 进度条 def progress_bar(iteration, total, length=50): percent = int(100 * (iteration / total)) filled = int(length * iteration // total) bar = '█' * filled + '-' * (length - filled) print(f'\r{Colors.CYAN}|{bar}| {percent}% {Colors.END}', end='\r') if iteration == total: print() # 使用进度条 for i in range(101): progress_bar(i, 100) import time time.sleep(0.01)

4. 系统日志

python

import logging import logging.handlers import sys # 4.1 配置日志 def setup_logger(name, level=logging.INFO): logger = logging.getLogger(name) logger.setLevel(level) # 控制台处理器 console_handler = logging.StreamHandler(sys.stdout) console_format = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) console_handler.setFormatter(console_format) # 文件处理器 file_handler = logging.FileHandler('app.log') file_format = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s' ) file_handler.setFormatter(file_format) # 添加处理器 logger.addHandler(console_handler) logger.addHandler(file_handler) return logger # 4.2 使用日志 logger = setup_logger('system_monitor') logger.info("System monitor started") logger.warning("High CPU usage detected") logger.error("Disk space critical") logger.debug("Debug information") # 4.3 轮转日志文件 rotating_handler = logging.handlers.RotatingFileHandler( 'app.log', maxBytes=1024*1024, backupCount=5 )

八、最佳实践与注意事项

1. 跨平台兼容性

python

import os import sys # 1.1 路径分隔符 if os.name == 'nt': # Windows path_sep = '\\' else: # Unix/Linux/Mac path_sep = '/' # 更好的方式是使用os.path.join path = os.path.join('dir', 'subdir', 'file.txt') # 1.2 换行符处理 text = "Line1\nLine2\nLine3" if sys.platform == 'win32': text = text.replace('\n', '\r\n') # 1.3 可执行文件扩展名 if sys.platform == 'win32': python_exe = 'python.exe' else: python_exe = 'python3' # 1.4 临时文件处理 import tempfile # 创建临时文件（自动清理） with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp: tmp.write('Temporary content') tmp_path = tmp.name # 使用后清理 os.unlink(tmp_path) # 创建临时目录 with tempfile.TemporaryDirectory() as tmpdir: print(f"Using temp dir: {tmpdir}") # 临时目录会在with块结束后自动删除

2. 错误处理与异常

python

import os import errno # 2.1 文件操作错误处理 try: with open('nonexistent.txt', 'r') as f: content = f.read() except FileNotFoundError: print("File not found") except PermissionError: print("Permission denied") except IOError as e: print(f"I/O error: {e}") except Exception as e: print(f"Unexpected error: {e}") # 2.2 检查文件是否存在（避免竞争条件） def safe_read_file(filename): try: with open(filename, 'r') as f: return f.read() except FileNotFoundError: return None # 2.3 原子文件操作 import tempfile import shutil def atomic_write(filename, content): """原子写入文件，避免写入过程中崩溃导致文件损坏""" temp_file = None try: # 创建临时文件 with tempfile.NamedTemporaryFile( mode='w', dir=os.path.dirname(filename), delete=False ) as tmp: temp_file = tmp.name tmp.write(content) # 原子重命名 os.replace(temp_file, filename) finally: # 清理临时文件（如果重命名失败） if temp_file and os.path.exists(temp_file): os.unlink(temp_file)

3. 性能优化建议

python

import os import timeit from pathlib import Path # 3.1 批量文件操作使用listdir # 慢的方式 files = [] for item in os.listdir('.'): if os.path.isfile(item): files.append(item) # 快的方式（使用列表推导式） files = [item for item in os.listdir('.') if os.path.isfile(item)] # 3.2 避免重复的stat调用 def get_file_info_bad(filepath): """不好的实现：多次调用stat""" return { 'exists': os.path.exists(filepath), 'size': os.path.getsize(filepath) if os.path.exists(filepath) else 0, 'mtime': os.path.getmtime(filepath) if os.path.exists(filepath) else 0 } def get_file_info_good(filepath): """好的实现：只调用一次stat""" try: stat_info = os.stat(filepath) return { 'exists': True, 'size': stat_info.st_size, 'mtime': stat_info.st_mtime } except FileNotFoundError: return { 'exists': False, 'size': 0, 'mtime': 0 } # 3.3 使用pathlib提高代码可读性 # 传统方式 import os def find_py_files_old(directory): py_files = [] for root, dirs, files in os.walk(directory): for file in files: if file.endswith('.py'): py_files.append(os.path.join(root, file)) return py_files # pathlib方式 from pathlib import Path def find_py_files_new(directory): path = Path(directory) return list(path.rglob('*.py'))

九、实战应用示例

1. 系统监控脚本

python

#!/usr/bin/env python3 """ 系统监控脚本：监控CPU、内存、磁盘使用情况 """ import psutil import time import logging from datetime import datetime class SystemMonitor: def __init__(self, alert_thresholds=None): self.thresholds = alert_thresholds or { 'cpu_percent': 80, 'memory_percent': 85, 'disk_percent': 90 } self.setup_logging() def setup_logging(self): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('system_monitor.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def check_cpu(self): cpu_percent = psutil.cpu_percent(interval=1) if cpu_percent > self.thresholds['cpu_percent']: self.logger.warning(f"High CPU usage: {cpu_percent}%") return cpu_percent def check_memory(self): memory = psutil.virtual_memory() if memory.percent > self.thresholds['memory_percent']: self.logger.warning(f"High memory usage: {memory.percent}%") return memory.percent def check_disk(self, path='/'): try: disk = psutil.disk_usage(path) if disk.percent > self.thresholds['disk_percent']: self.logger.warning(f"High disk usage on {path}: {disk.percent}%") return disk.percent except Exception as e: self.logger.error(f"Error checking disk {path}: {e}") return None def check_processes(self, process_name=None): processes = [] for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']): try: info = proc.info if process_name and process_name.lower() in info['name'].lower(): processes.append(info) elif info['cpu_percent'] > 10 or info['memory_percent'] > 5: processes.append(info) except (psutil.NoSuchProcess, psutil.AccessDenied): continue return processes def generate_report(self): report = { 'timestamp': datetime.now().isoformat(), 'cpu': self.check_cpu(), 'memory': self.check_memory(), 'disk': self.check_disk(), 'top_processes': self.check_processes()[:5] } return report def monitor(self, interval=60, duration=3600): """监控系统指定时间""" end_time = time.time() + duration self.logger.info(f"Starting system monitor for {duration} seconds") while time.time() < end_time: try: report = self.generate_report() self.logger.info(f"System status: CPU={report['cpu']}%, " f"Memory={report['memory']}%, " f"Disk={report['disk']}%") if time.time() + interval < end_time: time.sleep(interval) except KeyboardInterrupt: self.logger.info("Monitoring stopped by user") break except Exception as e: self.logger.error(f"Monitoring error: {e}") time.sleep(interval) if __name__ == '__main__': monitor = SystemMonitor() monitor.monitor(interval=300, duration=3600) # 每5分钟检查一次，持续1小时

2. 批量文件处理器

python

#!/usr/bin/env python3 """ 批量文件处理器：重命名、转换、整理文件 """ import os import shutil from pathlib import Path import hashlib import argparse class BatchFileProcessor: def __init__(self, source_dir, target_dir=None): self.source_dir = Path(source_dir).resolve() self.target_dir = Path(target_dir).resolve() if target_dir else self.source_dir # 确保目标目录存在 self.target_dir.mkdir(parents=True, exist_ok=True) def get_file_hash(self, filepath, algorithm='md5'): """计算文件哈希值""" hash_func = getattr(hashlib, algorithm)() with open(filepath, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hash_func.update(chunk) return hash_func.hexdigest() def find_duplicates(self): """查找重复文件""" hashes = {} duplicates = [] for filepath in self.source_dir.rglob('*'): if filepath.is_file(): file_hash = self.get_file_hash(filepath) if file_hash in hashes: duplicates.append((filepath, hashes[file_hash])) else: hashes[file_hash] = filepath return duplicates def organize_by_extension(self): """按扩展名整理文件""" for filepath in self.source_dir.rglob('*'): if filepath.is_file(): # 获取扩展名（不含点） ext = filepath.suffix[1:].lower() if filepath.suffix else 'no_extension' # 创建目标目录 target_ext_dir = self.target_dir / ext target_ext_dir.mkdir(exist_ok=True) # 移动文件 target_path = target_ext_dir / filepath.name if not target_path.exists(): shutil.move(str(filepath), str(target_path)) print(f"Moved: {filepath.name} -> {ext}/") def rename_sequential(self, prefix='file', start=1): """顺序重命名文件""" counter = start for filepath in sorted(self.source_dir.rglob('*')): if filepath.is_file(): # 保留原扩展名 new_name = f"{prefix}_{counter:04d}{filepath.suffix}" new_path = filepath.parent / new_name if not new_path.exists(): filepath.rename(new_path) print(f"Renamed: {filepath.name} -> {new_name}") counter += 1 def convert_text_encoding(self, source_encoding='utf-8', target_encoding='utf-8'): """转换文本文件编码""" converted = 0 for filepath in self.source_dir.rglob('*.txt'): try: # 读取原文件 with open(filepath, 'r', encoding=source_encoding) as f: content = f.read() # 写入新编码 with open(filepath, 'w', encoding=target_encoding) as f: f.write(content) converted += 1 print(f"Converted: {filepath.name}") except UnicodeDecodeError: print(f"Skipped (wrong encoding): {filepath.name}") except Exception as e: print(f"Error converting {filepath.name}: {e}") return converted def backup_files(self, backup_dir, extensions=None): """备份指定扩展名的文件""" backup_path = Path(backup_dir).resolve() backup_path.mkdir(parents=True, exist_ok=True) backed_up = 0 for filepath in self.source_dir.rglob('*'): if filepath.is_file(): # 检查扩展名 if extensions and filepath.suffix.lower() not in extensions: continue # 创建目标路径 rel_path = filepath.relative_to(self.source_dir) target_path = backup_path / rel_path # 确保目标目录存在 target_path.parent.mkdir(parents=True, exist_ok=True) # 复制文件 shutil.copy2(filepath, target_path) backed_up += 1 return backed_up def main(): parser = argparse.ArgumentParser(description='批量文件处理器') parser.add_argument('source', help='源目录') parser.add_argument('--target', help='目标目录') parser.add_argument('--organize', action='store_true', help='按扩展名整理') parser.add_argument('--rename', action='store_true', help='顺序重命名') parser.add_argument('--backup', help='备份目录') parser.add_argument('--extensions', nargs='+', help='备份的文件扩展名') args = parser.parse_args() processor = BatchFileProcessor(args.source, args.target) if args.organize: print("按扩展名整理文件...") processor.organize_by_extension() if args.rename: print("顺序重命名文件...") processor.rename_sequential() if args.backup: print(f"备份文件到 {args.backup}...") count = processor.backup_files(args.backup, args.extensions) print(f"备份了 {count} 个文件") # 查找重复文件 duplicates = processor.find_duplicates() if duplicates: print(f"发现 {len(duplicates)} 组重复文件:") for dup in duplicates[:5]: # 只显示前5组 print(f" {dup[0].name} 重复于 {dup[1].name}") if __name__ == '__main__': main()

十、总结

本文详细介绍了Python中常用的操作系统命令，涵盖了文件操作、目录管理、进程控制、系统监控等多个方面。以下是关键要点总结：

1. 文件操作：使用open()进行文件读写，shutil进行文件复制移动，os.path进行文件属性查询

2. 目录管理：使用os.makedirs()创建目录，os.walk()遍历目录树，pathlib提供面向对象路径操作

3. 进程控制：使用subprocess执行外部命令，multiprocessing进行多进程编程

4. 系统监控：使用psutil获取系统资源信息，platform获取系统平台信息

5. 环境变量：通过os.environ访问和修改环境变量

6. 跨平台兼容：注意不同操作系统的路径分隔符、换行符等差异

最佳实践建议：

• 使用pathlib代替传统的os.path操作，代码更简洁

• 处理文件时始终使用上下文管理器（with语句）

• 考虑跨平台兼容性，避免硬编码路径分隔符

• 对大文件或大量文件操作时注意性能优化

• 正确处理异常，特别是文件操作可能引发的各种错误

通过掌握这些操作系统命令，您将能够编写出强大、健壮且高效的Python脚本，轻松应对各种系统管理和自动化任务。