系统和进程管理

目录

系统和进程管理概述

系统和进程管理是Python在系统编程领域的重要应用，允许我们与操作系统进行交互、管理进程和线程、获取系统信息等。这对于系统管理员、DevOps工程师和需要与操作系统深度集成的开发者来说非常重要。

系统管理的重要性

  1. 自动化运维：自动化系统管理任务
  2. 进程监控：监控和管理运行中的进程
  3. 资源管理：管理系统资源和性能
  4. 文件操作：高级文件系统操作
  5. 环境配置：管理系统环境和配置

os模块

os模块提供了与操作系统交互的功能。

基本操作系统功能

import os
import sys

# 获取操作系统信息
print("操作系统信息:")
print(f"  操作系统: {os.name}")
print(f"  平台: {sys.platform}")
print(f"  Python版本: {sys.version}")

# 当前工作目录
print(f"\n当前工作目录: {os.getcwd()}")

# 列出目录内容
print(f"\n当前目录内容:")
try:
    contents = os.listdir('.')
    for item in contents[:10]:  # 只显示前10项
        print(f"  {item}")
    if len(contents) > 10:
        print(f"  ... 还有 {len(contents) - 10} 项")
except OSError as e:
    print(f"  无法列出目录内容: {e}")

# 环境变量
print(f"\n环境变量:")
print(f"  PATH: {os.environ.get('PATH', '未设置')[:100]}...")
print(f"  HOME: {os.environ.get('HOME', os.environ.get('USERPROFILE', '未设置'))}")
print(f"  USERNAME: {os.environ.get('USERNAME', os.environ.get('USER', '未设置'))}")

目录和文件操作

import os
import tempfile
from pathlib import Path

# 目录操作
print("目录操作:")
try:
    # 创建临时目录进行测试
    with tempfile.TemporaryDirectory() as temp_dir:
        print(f"  临时目录: {temp_dir}")
        
        # 创建子目录
        test_dir = os.path.join(temp_dir, "test_dir")
        os.mkdir(test_dir)
        print(f"  创建目录: {test_dir}")
        
        # 创建多级目录
        nested_dir = os.path.join(temp_dir, "parent", "child", "grandchild")
        os.makedirs(nested_dir, exist_ok=True)
        print(f"  创建多级目录: {nested_dir}")
        
        # 切换目录
        original_dir = os.getcwd()
        os.chdir(temp_dir)
        print(f"  切换到: {os.getcwd()}")
        
        # 切换回原目录
        os.chdir(original_dir)
        print(f"  切换回: {os.getcwd()}")
        
        # 删除目录
        os.rmdir(test_dir)
        print(f"  删除目录: {test_dir}")
        
except OSError as e:
    print(f"  目录操作错误: {e}")

# 文件操作
print(f"\n文件操作:")
try:
    with tempfile.NamedTemporaryFile(delete=False) as temp_file:
        temp_file_path = temp_file.name
        temp_file.write(b"Hello, World!")
    
    print(f"  创建临时文件: {temp_file_path}")
    
    # 检查文件状态
    stat_info = os.stat(temp_file_path)
    print(f"  文件大小: {stat_info.st_size} 字节")
    print(f"  创建时间: {stat_info.st_ctime}")
    print(f"  修改时间: {stat_info.st_mtime}")
    
    # 重命名文件
    new_path = temp_file_path + ".renamed"
    os.rename(temp_file_path, new_path)
    print(f"  重命名文件: {new_path}")
    
    # 删除文件
    os.remove(new_path)
    print(f"  删除文件: {new_path}")
    
except OSError as e:
    print(f"  文件操作错误: {e}")

路径操作

import os
from pathlib import Path

# 路径操作
print("路径操作:")
current_file = __file__
print(f"  当前文件: {current_file}")

# 路径分解
print(f"  目录部分: {os.path.dirname(current_file)}")
print(f"  文件名: {os.path.basename(current_file)}")
print(f"  文件名(无扩展名): {os.path.splitext(os.path.basename(current_file))[0]}")
print(f"  扩展名: {os.path.splitext(current_file)[1]}")

# 路径组合
base_dir = "/home/user"
filename = "document.txt"
full_path = os.path.join(base_dir, filename)
print(f"  路径组合: {full_path}")

# 路径检查
print(f"  路径存在: {os.path.exists(current_file)}")
print(f"  是文件: {os.path.isfile(current_file)}")
print(f"  是目录: {os.path.isdir(os.path.dirname(current_file))}")
print(f"  是绝对路径: {os.path.isabs(current_file)}")

# 获取绝对路径
relative_path = "./test.txt"
absolute_path = os.path.abspath(relative_path)
print(f"  相对路径 '{relative_path}' 的绝对路径: {absolute_path}")

# 使用pathlib进行现代路径操作
print(f"\n使用pathlib:")
path = Path(current_file)
print(f"  当前文件: {path}")
print(f"  父目录: {path.parent}")
print(f"  文件名: {path.name}")
print(f"  文件stem: {path.stem}")
print(f"  扩展名: {path.suffix}")
print(f"  是否存在: {path.exists()}")
print(f"  是否是文件: {path.is_file()}")

sys模块

sys模块提供了与Python解释器交互的功能。

系统相关参数

import sys
import os

# Python解释器信息
print("Python解释器信息:")
print(f"  Python版本: {sys.version}")
print(f"  Python版本信息: {sys.version_info}")
print(f"  Python实现: {sys.implementation}")
print(f"  可执行文件路径: {sys.executable}")

# 系统参数
print(f"\n系统参数:")
print(f"  平台: {sys.platform}")
print(f"  字节序: {sys.byteorder}")
print(f"  最大递归深度: {sys.getrecursionlimit()}")

# 命令行参数
print(f"\n命令行参数:")
print(f"  参数列表: {sys.argv}")
for i, arg in enumerate(sys.argv):
    print(f"  参数 {i}: {arg}")

# 模块搜索路径
print(f"\n模块搜索路径:")
for i, path in enumerate(sys.path[:5]):  # 只显示前5个
    print(f"  {i}: {path}")
if len(sys.path) > 5:
    print(f"  ... 还有 {len(sys.path) - 5} 个路径")

# 标准流
print(f"\n标准流:")
print(f"  标准输入: {sys.stdin}")
print(f"  标准输出: {sys.stdout}")
print(f"  标准错误: {sys.stderr}")

# 内存使用信息
print(f"\n内存使用信息:")
print(f"  已分配内存块数: {sys.getallocatedblocks()}")

系统退出和异常处理

import sys

# 系统退出
print("系统退出示例:")
print("  sys.exit() - 正常退出")
print("  sys.exit(1) - 异常退出")
print("  sys.exit('错误信息') - 带信息退出")

# 获取异常信息
print(f"\n异常信息:")
try:
    1 / 0
except ZeroDivisionError:
    print(f"  最后异常类型: {sys.exc_info()[0]}")
    print(f"  最后异常值: {sys.exc_info()[1]}")
    print(f"  最后异常追踪: {sys.exc_info()[2]}")

# 标准流重定向
print(f"\n标准流操作:")
original_stdout = sys.stdout
# 重定向输出到文件
# with open('output.txt', 'w') as f:
#     sys.stdout = f
#     print("这会写入文件")
# sys.stdout = original_stdout
# print("这会输出到控制台")

subprocess模块

subprocess模块用于创建和管理子进程。

基本子进程操作

import subprocess
import sys

# 执行简单命令
print("基本子进程操作:")
try:
    # 执行命令并获取输出
    if sys.platform.startswith('win'):
        result = subprocess.run(['cmd', '/c', 'echo', 'Hello, World!'], 
                              capture_output=True, text=True, check=True)
    else:
        result = subprocess.run(['echo', 'Hello, World!'], 
                              capture_output=True, text=True, check=True)
    
    print(f"  命令输出: {result.stdout.strip()}")
    print(f"  返回码: {result.returncode}")
    
except subprocess.CalledProcessError as e:
    print(f"  命令执行失败: {e}")
except FileNotFoundError:
    print("  命令未找到")

# 执行系统命令
print(f"\n执行系统命令:")
try:
    if sys.platform.startswith('win'):
        # Windows命令
        result = subprocess.run(['dir'], shell=True, capture_output=True, text=True)
        print("  Windows目录列表命令执行成功" if result.returncode == 0 else "  命令执行失败")
    else:
        # Unix/Linux命令
        result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
        if result.returncode == 0:
            print("  Unix目录列表:")
            print(result.stdout[:200] + "..." if len(result.stdout) > 200 else result.stdout)
        else:
            print("  命令执行失败")
            
except Exception as e:
    print(f"  命令执行错误: {e}")

高级子进程操作

import subprocess
import sys
import time

# 实时输出
print("实时输出:")
try:
    if sys.platform.startswith('win'):
        process = subprocess.Popen(['ping', '127.0.0.1', '-n', '3'], 
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, 
                                 text=True, shell=True)
    else:
        process = subprocess.Popen(['ping', '127.0.0.1', '-c', '3'], 
                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE, 
                                 text=True)
    
    # 实时读取输出
    while True:
        output = process.stdout.readline()
        if output == '' and process.poll() is not None:
            break
        if output:
            print(f"  {output.strip()}")
    
    # 等待进程结束
    process.wait()
    print(f"  进程返回码: {process.returncode}")
    
except Exception as e:
    print(f"  实时输出错误: {e}")

# 进程间通信
print(f"\n进程间通信:")
try:
    # 创建子进程并传递输入
    if sys.platform.startswith('win'):
        process = subprocess.Popen(['findstr', 'test'], 
                                 stdin=subprocess.PIPE, 
                                 stdout=subprocess.PIPE, 
                                 stderr=subprocess.PIPE, 
                                 text=True, shell=True)
    else:
        process = subprocess.Popen(['grep', 'test'], 
                                 stdin=subprocess.PIPE, 
                                 stdout=subprocess.PIPE, 
                                 stderr=subprocess.PIPE, 
                                 text=True)
    
    # 向子进程发送输入
    input_text = "this is a test\nanother line\nfinal test line\n"
    stdout, stderr = process.communicate(input=input_text)
    
    print(f"  输入文本: {input_text.strip()}")
    print(f"  匹配结果: {stdout.strip()}")
    print(f"  错误输出: {stderr.strip()}")
    print(f"  返回码: {process.returncode}")
    
except Exception as e:
    print(f"  进程间通信错误: {e}")

进程管理

import subprocess
import time
import signal
import sys

# 启动长时间运行的进程
print("长时间运行进程:")
try:
    if sys.platform.startswith('win'):
        # Windows: 启动记事本
        process = subprocess.Popen(['notepad.exe'], 
                                 stdout=subprocess.PIPE, 
                                 stderr=subprocess.PIPE)
    else:
        # Unix/Linux: 启动sleep命令
        process = subprocess.Popen(['sleep', '10'], 
                                 stdout=subprocess.PIPE, 
                                 stderr=subprocess.PIPE)
    
    print(f"  启动进程 PID: {process.pid}")
    print(f"  进程是否运行: {process.poll() is None}")
    
    # 等待一段时间
    time.sleep(2)
    
    # 检查进程状态
    if process.poll() is None:
        print("  进程仍在运行")
        # 终止进程
        process.terminate()
        print("  发送终止信号")
        
        # 等待进程结束
        try:
            process.wait(timeout=5)
            print(f"  进程已结束，返回码: {process.returncode}")
        except subprocess.TimeoutExpired:
            print("  进程未响应，强制终止")
            process.kill()
            process.wait()
            print(f"  进程已强制终止，返回码: {process.returncode}")
    else:
        print(f"  进程已结束，返回码: {process.poll()}")
        
except Exception as e:
    print(f"  进程管理错误: {e}")

进程管理

Python提供了多种方式进行进程管理。

multiprocessing模块

import multiprocessing
import time
import os

# 基本多进程
def worker_function(name, duration):
    """工作进程函数"""
    print(f"进程 {name} (PID: {os.getpid()}) 开始工作")
    time.sleep(duration)
    print(f"进程 {name} 完成工作")
    return f"进程 {name} 的结果"

def multiprocessing_demo():
    """多进程演示"""
    print("多进程演示:")
    
    # 创建进程
    processes = []
    for i in range(3):
        process = multiprocessing.Process(target=worker_function, 
                                       args=(f"Worker-{i}", 2))
        processes.append(process)
        process.start()
        print(f"  启动进程 {i} (PID: {process.pid})")
    
    # 等待所有进程完成
    for i, process in enumerate(processes):
        process.join()
        print(f"  进程 {i} 已完成")
    
    print("所有进程完成")

# 进程池
def square_number(n):
    """计算平方"""
    time.sleep(0.1)  # 模拟计算时间
    return n * n

def process_pool_demo():
    """进程池演示"""
    print("\n进程池演示:")
    
    numbers = list(range(10))
    print(f"  计算数字: {numbers}")
    
    # 使用进程池
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(square_number, numbers)
    
    print(f"  计算结果: {results}")

# 进程间通信
def producer(queue):
    """生产者进程"""
    for i in range(5):
        item = f"item-{i}"
        queue.put(item)
        print(f"生产者: 放入 {item}")
        time.sleep(0.5)
    queue.put(None)  # 发送结束信号

def consumer(queue):
    """消费者进程"""
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"消费者: 获取 {item}")
        time.sleep(0.3)
    print("消费者: 完成")

def ipc_demo():
    """进程间通信演示"""
    print("\n进程间通信演示:")
    
    # 创建队列
    queue = multiprocessing.Queue()
    
    # 创建生产者和消费者进程
    producer_process = multiprocessing.Process(target=producer, args=(queue,))
    consumer_process = multiprocessing.Process(target=consumer, args=(queue,))
    
    # 启动进程
    producer_process.start()
    consumer_process.start()
    
    # 等待进程完成
    producer_process.join()
    consumer_process.join()
    
    print("进程间通信完成")

# 运行多进程示例
if __name__ == '__main__':
    multiprocessing_demo()
    process_pool_demo()
    ipc_demo()

concurrent.futures模块

import concurrent.futures
import time
import requests

# 使用ThreadPoolExecutor
def fetch_url(url):
    """获取URL内容"""
    try:
        response = requests.get(url, timeout=5)
        return f"{url}: {response.status_code} ({len(response.content)} 字节)"
    except Exception as e:
        return f"{url}: 错误 - {e}"

def thread_pool_demo():
    """线程池演示"""
    print("线程池演示:")
    
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2'
    ]
    
    print("  请求URL列表:")
    for url in urls:
        print(f"    {url}")
    
    # 使用线程池并发执行
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        # 提交任务
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        
        # 获取结果
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                result = future.result()
                print(f"  {result}")
            except Exception as e:
                print(f"  {url} 产生异常: {e}")

# 使用ProcessPoolExecutor
def cpu_intensive_task(n):
    """CPU密集型任务"""
    result = 0
    for i in range(n):
        result += i * i
    return result

def process_pool_demo():
    """进程池演示"""
    print("\n进程池演示:")
    
    tasks = [1000000, 2000000, 1500000, 1200000]
    print(f"  计算任务: {tasks}")
    
    # 使用进程池并行计算
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        # 提交任务
        future_to_task = {executor.submit(cpu_intensive_task, task): task for task in tasks}
        
        # 获取结果
        results = []
        for future in concurrent.futures.as_completed(future_to_task):
            task = future_to_task[future]
            try:
                result = future.result()
                results.append((task, result))
                print(f"  任务 {task} 完成")
            except Exception as e:
                print(f"  任务 {task} 产生异常: {e}")
        
        # 按任务大小排序结果
        results.sort(key=lambda x: x[0])
        print(f"  计算结果: {[result for _, result in results]}")

# 运行并发示例
# thread_pool_demo()  # 需要网络连接
process_pool_demo()

线程管理

Python提供了多种线程管理方式。

threading模块

import threading
import time
import queue

# 基本线程操作
def worker_thread(name, duration):
    """工作线程函数"""
    print(f"线程 {name} 开始工作")
    time.sleep(duration)
    print(f"线程 {name} 完成工作")

def threading_demo():
    """线程演示"""
    print("线程演示:")
    
    # 创建线程
    threads = []
    for i in range(3):
        thread = threading.Thread(target=worker_thread, args=(f"Thread-{i}", 2))
        threads.append(thread)
        thread.start()
        print(f"  启动线程 {i}")
    
    # 等待所有线程完成
    for i, thread in enumerate(threads):
        thread.join()
        print(f"  线程 {i} 已完成")
    
    print("所有线程完成")

# 线程同步
class Counter:
    """计数器类"""
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()
    
    def increment(self):
        """增加计数"""
        with self.lock:
            temp = self.value
            time.sleep(0.0001)  # 模拟处理时间
            self.value = temp + 1

def counter_worker(counter, iterations):
    """计数器工作线程"""
    for _ in range(iterations):
        counter.increment()

def thread_synchronization_demo():
    """线程同步演示"""
    print("\n线程同步演示:")
    
    counter = Counter()
    threads = []
    
    # 创建多个线程同时修改计数器
    for i in range(5):
        thread = threading.Thread(target=counter_worker, args=(counter, 100))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()
    
    print(f"  期望值: 500")
    print(f"  实际值: {counter.value}")
    print(f"  结果{'正确' if counter.value == 500 else '错误'}")

# 线程间通信
def producer_thread(q):
    """生产者线程"""
    for i in range(5):
        item = f"item-{i}"
        q.put(item)
        print(f"生产者: 放入 {item}")
        time.sleep(0.5)
    q.put(None)  # 发送结束信号

def consumer_thread(q):
    """消费者线程"""
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费者: 获取 {item}")
        time.sleep(0.3)
    print("消费者: 完成")

def thread_ipc_demo():
    """线程间通信演示"""
    print("\n线程间通信演示:")
    
    # 创建队列
    q = queue.Queue()
    
    # 创建生产者和消费者线程
    producer = threading.Thread(target=producer_thread, args=(q,))
    consumer = threading.Thread(target=consumer_thread, args=(q,))
    
    # 启动线程
    producer.start()
    consumer.start()
    
    # 等待线程完成
    producer.join()
    consumer.join()
    
    print("线程间通信完成")

# 运行线程示例
threading_demo()
thread_synchronization_demo()
thread_ipc_demo()

系统信息获取

获取系统信息对于系统管理和监控非常重要。

psutil库

# 需要安装psutil: pip install psutil
try:
    import psutil
    import datetime
    
    print("系统信息获取:")
    
    # CPU信息
    print("CPU信息:")
    print(f"  CPU逻辑核心数: {psutil.cpu_count()}")
    print(f"  CPU物理核心数: {psutil.cpu_count(logical=False)}")
    print(f"  CPU使用率: {psutil.cpu_percent(interval=1)}%")
    
    # 内存信息
    print("\n内存信息:")
    memory = psutil.virtual_memory()
    print(f"  总内存: {memory.total / (1024**3):.2f} GB")
    print(f"  可用内存: {memory.available / (1024**3):.2f} GB")
    print(f"  内存使用率: {memory.percent}%")
    
    # 磁盘信息
    print("\n磁盘信息:")
    disk = psutil.disk_usage('/')
    print(f"  总空间: {disk.total / (1024**3):.2f} GB")
    print(f"  已使用: {disk.used / (1024**3):.2f} GB")
    print(f"  可用空间: {disk.free / (1024**3):.2f} GB")
    print(f"  使用率: {disk.percent}%")
    
    # 网络信息
    print("\n网络信息:")
    net_io = psutil.net_io_counters()
    print(f"  发送字节: {net_io.bytes_sent / (1024**2):.2f} MB")
    print(f"  接收字节: {net_io.bytes_recv / (1024**2):.2f} MB")
    
    # 进程信息
    print("\n进程信息:")
    print(f"  进程总数: {len(psutil.pids())}")
    
    # 当前进程信息
    current_process = psutil.Process()
    print(f"  当前进程ID: {current_process.pid}")
    print(f"  当前进程名称: {current_process.name()}")
    print(f"  当前进程状态: {current_process.status()}")
    print(f"  当前进程内存使用: {current_process.memory_info().rss / (1024**2):.2f} MB")
    
    # 系统启动时间
    boot_time = datetime.datetime.fromtimestamp(psutil.boot_time())
    print(f"\n系统信息:")
    print(f"  系统启动时间: {boot_time}")
    print(f"  系统运行时间: {datetime.datetime.now() - boot_time}")
    
except ImportError:
    print("请安装psutil库: pip install psutil")

使用标准库获取系统信息

import os
import platform
import sys
import time

# 使用标准库获取系统信息
print("使用标准库获取系统信息:")

# 平台信息
print("平台信息:")
print(f"  系统: {platform.system()}")
print(f"  节点名称: {platform.node()}")
print(f"  发布版本: {platform.release()}")
print(f"  版本: {platform.version()}")
print(f"  机器: {platform.machine()}")
print(f"  处理器: {platform.processor()}")
print(f"  架构: {platform.architecture()}")

# Python信息
print(f"\nPython信息:")
print(f"  Python实现: {platform.python_implementation()}")
print(f"  Python版本: {platform.python_version()}")
print(f"  Python编译器: {platform.python_compiler()}")

# 环境信息
print(f"\n环境信息:")
print(f"  当前工作目录: {os.getcwd()}")
print(f"  用户: {os.environ.get('USERNAME', os.environ.get('USER', '未知'))}")
print(f"  主目录: {os.path.expanduser('~')}")

# 时间信息
print(f"\n时间信息:")
print(f"  当前时间: {time.ctime()}")
print(f"  时间戳: {time.time()}")

文件系统操作

高级文件系统操作对于系统管理非常重要。

shutil模块

import shutil
import os
import tempfile
from pathlib import Path

# 文件和目录操作
print("文件系统操作:")

try:
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)
        print(f"  临时目录: {temp_path}")
        
        # 创建测试文件
        test_file = temp_path / "test.txt"
        test_file.write_text("Hello, World!")
        print(f"  创建文件: {test_file}")
        
        # 复制文件
        copied_file = temp_path / "test_copy.txt"
        shutil.copy2(test_file, copied_file)
        print(f"  复制文件: {copied_file}")
        
        # 创建测试目录
        test_dir = temp_path / "test_dir"
        test_dir.mkdir()
        (test_dir / "sub_file.txt").write_text("Sub file content")
        print(f"  创建目录: {test_dir}")
        
        # 复制目录
        copied_dir = temp_path / "test_dir_copy"
        shutil.copytree(test_dir, copied_dir)
        print(f"  复制目录: {copied_dir}")
        
        # 移动文件
        moved_file = temp_path / "test_moved.txt"
        shutil.move(test_file, moved_file)
        print(f"  移动文件: {moved_file}")
        
        # 获取磁盘使用情况
        total, used, free = shutil.disk_usage(temp_dir)
        print(f"  磁盘使用情况:")
        print(f"    总空间: {total / (1024**3):.2f} GB")
        print(f"    已使用: {used / (1024**3):.2f} GB")
        print(f"    可用空间: {free / (1024**3):.2f} GB")
        
except Exception as e:
    print(f"  文件系统操作错误: {e}")

# 归档操作
print(f"\n归档操作:")
try:
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)
        
        # 创建测试文件
        for i in range(3):
            (temp_path / f"file_{i}.txt").write_text(f"Content of file {i}")
        
        # 创建归档
        archive_name = str(temp_path / "test_archive")
        shutil.make_archive(archive_name, 'zip', temp_dir)
        print(f"  创建ZIP归档: {archive_name}.zip")
        
        # 解压归档
        extract_dir = temp_path / "extracted"
        shutil.unpack_archive(f"{archive_name}.zip", extract_dir)
        print(f"  解压归档到: {extract_dir}")
        
except Exception as e:
    print(f"  归档操作错误: {e}")

glob和fnmatch模块

import glob
import fnmatch
import os
import tempfile
from pathlib import Path

# 文件模式匹配
print("文件模式匹配:")

try:
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)
        
        # 创建测试文件
        test_files = [
            "test1.txt", "test2.txt", "test3.log",
            "data1.csv", "data2.csv", "image1.png", "image2.jpg"
        ]
        
        for filename in test_files:
            (temp_path / filename).touch()
        
        # 使用glob匹配
        print("  glob匹配:")
        txt_files = glob.glob(str(temp_path / "*.txt"))
        print(f"    TXT文件: {len(txt_files)} 个")
        
        all_files = glob.glob(str(temp_path / "*.*"))
        print(f"    所有文件: {len(all_files)} 个")
        
        # 递归匹配
        sub_dir = temp_path / "subdir"
        sub_dir.mkdir()
        (sub_dir / "nested.txt").touch()
        
        recursive_files = glob.glob(str(temp_path / "**" / "*.txt"), recursive=True)
        print(f"    递归TXT文件: {len(recursive_files)} 个")
        
        # 使用fnmatch匹配
        print("  fnmatch匹配:")
        patterns = ["*.txt", "*.csv", "test*"]
        for pattern in patterns:
            matches = [f for f in test_files if fnmatch.fnmatch(f, pattern)]
            print(f"    {pattern}: {len(matches)} 个匹配")
            
except Exception as e:
    print(f"  文件模式匹配错误: {e}")

环境变量管理

环境变量管理是系统配置的重要部分。

环境变量操作

import os
import tempfile

# 环境变量管理
print("环境变量管理:")

# 查看环境变量
print("  当前环境变量:")
important_vars = ['PATH', 'HOME', 'USER', 'USERNAME', 'SHELL', 'TERM']
for var in important_vars:
    value = os.environ.get(var, '未设置')
    if var == 'PATH':
        print(f"    {var}: {value[:100]}..." if len(value) > 100 else f"    {var}: {value}")
    else:
        print(f"    {var}: {value}")

# 设置环境变量
print(f"\n  环境变量操作:")
original_value = os.environ.get('TEST_VAR', None)

# 临时设置环境变量
os.environ['TEST_VAR'] = 'test_value'
print(f"    设置 TEST_VAR = {os.environ['TEST_VAR']}")

# 使用environ上下文管理器（需要第三方库）
# from environ import Env
# env = Env()

# 恢复原始值
if original_value is not None:
    os.environ['TEST_VAR'] = original_value
else:
    os.environ.pop('TEST_VAR', None)

print(f"    恢复 TEST_VAR")

# 从文件加载环境变量
print(f"\n  从文件加载环境变量:")
try:
    with tempfile.NamedTemporaryFile(mode='w', suffix='.env', delete=False) as f:
        f.write("DATABASE_URL=postgresql://localhost/mydb\n")
        f.write("DEBUG=True\n")
        f.write("SECRET_KEY=mysecretkey\n")
        env_file = f.name
    
    print(f"    创建环境文件: {env_file}")
    
    # 读取环境文件
    with open(env_file, 'r') as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith('#'):
                if '=' in line:
                    key, value = line.split('=', 1)
                    print(f"    {key} = {value}")
    
    # 清理
    os.unlink(env_file)
    print(f"    删除环境文件: {env_file}")
    
except Exception as e:
    print(f"    环境文件操作错误: {e}")

配置管理

import os
import json
import configparser
from pathlib import Path

# 配置管理示例
class ConfigManager:
    """配置管理器"""
    
    def __init__(self, config_file="config.json"):
        self.config_file = Path(config_file)
        self.config = self.load_config()
    
    def load_config(self):
        """加载配置"""
        if self.config_file.exists():
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                print(f"加载配置文件错误: {e}")
                return self.get_default_config()
        else:
            return self.get_default_config()
    
    def save_config(self):
        """保存配置"""
        try:
            with open(self.config_file, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"保存配置文件错误: {e}")
    
    def get(self, key, default=None):
        """获取配置值"""
        return self.config.get(key, default)
    
    def set(self, key, value):
        """设置配置值"""
        self.config[key] = value
        self.save_config()
    
    def get_default_config(self):
        """获取默认配置"""
        return {
            "app_name": "MyApp",
            "version": "1.0.0",
            "debug": False,
            "database": {
                "host": "localhost",
                "port": 5432,
                "name": "mydb"
            },
            "logging": {
                "level": "INFO",
                "file": "app.log"
            }
        }

# INI配置文件处理
def ini_config_demo():
    """INI配置文件演示"""
    print("INI配置文件处理:")
    
    try:
        with tempfile.NamedTemporaryFile(mode='w', suffix='.ini', delete=False) as f:
            f.write("""
[app]
name = MyApp
version = 1.0.0
debug = False

[database]
host = localhost
port = 5432
name = mydb

[logging]
level = INFO
file = app.log
""")
            ini_file = f.name
        
        print(f"  创建INI文件: {ini_file}")
        
        # 读取INI配置
        config = configparser.ConfigParser()
        config.read(ini_file)
        
        print("  配置内容:")
        for section in config.sections():
            print(f"    [{section}]")
            for key, value in config.items(section):
                print(f"      {key} = {value}")
        
        # 清理
        os.unlink(ini_file)
        print(f"  删除INI文件: {ini_file}")
        
    except Exception as e:
        print(f"  INI配置文件操作错误: {e}")

# 演示配置管理
print("配置管理演示:")
config_manager = ConfigManager("demo_config.json")
print(f"  应用名称: {config_manager.get('app_name')}")
print(f"  数据库主机: {config_manager.get('database', {}).get('host')}")

# 修改配置
config_manager.set('debug', True)
print(f"  调试模式: {config_manager.get('debug')}")

# 清理演示文件
demo_config = Path("demo_config.json")
if demo_config.exists():
    demo_config.unlink()
    print(f"  删除演示配置文件: {demo_config}")

# INI配置演示
ini_config_demo()

实际应用示例

# 综合应用示例：系统监控和管理工具
import os
import sys
import time
import subprocess
import json
import threading
from datetime import datetime
from pathlib import Path

class SystemMonitor:
    """系统监控器"""
    
    def __init__(self, config_file="monitor_config.json"):
        self.config_file = Path(config_file)
        self.config = self.load_config()
        self.monitoring = False
        self.monitor_thread = None
    
    def load_config(self):
        """加载配置"""
        if self.config_file.exists():
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                print(f"加载配置错误: {e}")
        return self.get_default_config()
    
    def save_config(self):
        """保存配置"""
        try:
            with open(self.config_file, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"保存配置错误: {e}")
    
    def get_default_config(self):
        """获取默认配置"""
        return {
            "monitor_interval": 60,  # 监控间隔（秒）
            "log_file": "system_monitor.log",
            "alert_thresholds": {
                "cpu_usage": 80,
                "memory_usage": 85,
                "disk_usage": 90
            },
            "process_monitoring": {
                "enabled": True,
                "processes": []
            }
        }
    
    def get_system_info(self):
        """获取系统信息"""
        info = {
            "timestamp": datetime.now().isoformat(),
            "cpu_usage": self.get_cpu_usage(),
            "memory_usage": self.get_memory_usage(),
            "disk_usage": self.get_disk_usage(),
            "network_usage": self.get_network_usage()
        }
        return info
    
    def get_cpu_usage(self):
        """获取CPU使用率"""
        try:
            import psutil
            return psutil.cpu_percent(interval=1)
        except ImportError:
            # 使用系统命令
            try:
                if sys.platform.startswith('win'):
                    result = subprocess.run(['wmic', 'cpu', 'get', 'loadpercentage'], 
                                          capture_output=True, text=True)
                    lines = result.stdout.strip().split('\n')
                    if len(lines) > 1:
                        return int(lines[1].strip())
                else:
                    result = subprocess.run(['top', '-bn1'], 
                                          capture_output=True, text=True)
                    # 解析top输出获取CPU使用率
                    # 这里简化处理
                    return 0.0
            except:
                return 0.0
    
    def get_memory_usage(self):
        """获取内存使用率"""
        try:
            import psutil
            return psutil.virtual_memory().percent
        except ImportError:
            return 0.0
    
    def get_disk_usage(self):
        """获取磁盘使用率"""
        try:
            import psutil
            return psutil.disk_usage('/').percent
        except ImportError:
            return 0.0
    
    def get_network_usage(self):
        """获取网络使用情况"""
        try:
            import psutil
            net_io = psutil.net_io_counters()
            return {
                "bytes_sent": net_io.bytes_sent,
                "bytes_recv": net_io.bytes_recv
            }
        except ImportError:
            return {"bytes_sent": 0, "bytes_recv": 0}
    
    def check_alerts(self, info):
        """检查告警条件"""
        alerts = []
        thresholds = self.config.get("alert_thresholds", {})
        
        cpu_threshold = thresholds.get("cpu_usage", 80)
        if info["cpu_usage"] > cpu_threshold:
            alerts.append(f"CPU使用率过高: {info['cpu_usage']:.1f}%")
        
        memory_threshold = thresholds.get("memory_usage", 85)
        if info["memory_usage"] > memory_threshold:
            alerts.append(f"内存使用率过高: {info['memory_usage']:.1f}%")
        
        disk_threshold = thresholds.get("disk_usage", 90)
        if info["disk_usage"] > disk_threshold:
            alerts.append(f"磁盘使用率过高: {info['disk_usage']:.1f}%")
        
        return alerts
    
    def log_info(self, info, alerts=None):
        """记录信息到日志"""
        log_file = self.config.get("log_file", "system_monitor.log")
        try:
            with open(log_file, 'a', encoding='utf-8') as f:
                log_entry = {
                    "timestamp": info["timestamp"],
                    "cpu_usage": info["cpu_usage"],
                    "memory_usage": info["memory_usage"],
                    "disk_usage": info["disk_usage"]
                }
                if alerts:
                    log_entry["alerts"] = alerts
                
                f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
        except Exception as e:
            print(f"记录日志错误: {e}")
    
    def monitor_loop(self):
        """监控循环"""
        print("开始系统监控...")
        while self.monitoring:
            try:
                info = self.get_system_info()
                alerts = self.check_alerts(info)
                
                # 显示信息
                print(f"[{info['timestamp']}] "
                      f"CPU: {info['cpu_usage']:.1f}% "
                      f"内存: {info['memory_usage']:.1f}% "
                      f"磁盘: {info['disk_usage']:.1f}%")
                
                if alerts:
                    for alert in alerts:
                        print(f"  告警: {alert}")
                
                # 记录日志
                self.log_info(info, alerts)
                
                # 等待下次监控
                interval = self.config.get("monitor_interval", 60)
                time.sleep(interval)
                
            except KeyboardInterrupt:
                print("监控被中断")
                break
            except Exception as e:
                print(f"监控错误: {e}")
                time.sleep(10)  # 出错后等待10秒再继续
    
    def start_monitoring(self):
        """开始监控"""
        if not self.monitoring:
            self.monitoring = True
            self.monitor_thread = threading.Thread(target=self.monitor_loop)
            self.monitor_thread.daemon = True
            self.monitor_thread.start()
            print("系统监控已启动")
    
    def stop_monitoring(self):
        """停止监控"""
        if self.monitoring:
            self.monitoring = False
            if self.monitor_thread:
                self.monitor_thread.join(timeout=5)
            print("系统监控已停止")

class ProcessManager:
    """进程管理器"""
    
    def __init__(self):
        pass
    
    def list_processes(self):
        """列出进程"""
        try:
            import psutil
            processes = []
            for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
                try:
                    processes.append(proc.info)
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    pass
            return processes
        except ImportError:
            print("需要安装psutil库: pip install psutil")
            return []
    
    def kill_process(self, pid):
        """终止进程"""
        try:
            import psutil
            process = psutil.Process(pid)
            process.terminate()
            process.wait(timeout=3)
            return True
        except Exception as e:
            print(f"终止进程错误: {e}")
            return False
    
    def get_process_info(self, pid):
        """获取进程详细信息"""
        try:
            import psutil
            process = psutil.Process(pid)
            return {
                "pid": process.pid,
                "name": process.name(),
                "status": process.status(),
                "cpu_percent": process.cpu_percent(),
                "memory_percent": process.memory_percent(),
                "memory_info": process.memory_info()._asdict(),
                "create_time": datetime.fromtimestamp(process.create_time()).isoformat(),
                "cmdline": process.cmdline()
            }
        except Exception as e:
            print(f"获取进程信息错误: {e}")
            return None

# 系统管理工具
class SystemManager:
    """系统管理工具"""
    
    def __init__(self):
        self.monitor = SystemMonitor()
        self.process_manager = ProcessManager()
    
    def show_menu(self):
        """显示菜单"""
        print("\n=== 系统管理工具 ===")
        print("1. 系统监控")
        print("2. 进程管理")
        print("3. 系统信息")
        print("4. 配置管理")
        print("0. 退出")
        print("=" * 25)
    
    def system_monitor_menu(self):
        """系统监控菜单"""
        while True:
            print("\n--- 系统监控 ---")
            print("1. 启动监控")
            print("2. 停止监控")
            print("3. 查看当前状态")
            print("0. 返回主菜单")
            
            choice = input("请选择: ").strip()
            
            if choice == '1':
                self.monitor.start_monitoring()
            elif choice == '2':
                self.monitor.stop_monitoring()
            elif choice == '3':
                info = self.monitor.get_system_info()
                print(f"CPU使用率: {info['cpu_usage']:.1f}%")
                print(f"内存使用率: {info['memory_usage']:.1f}%")
                print(f"磁盘使用率: {info['disk_usage']:.1f}%")
            elif choice == '0':
                break
            else:
                print("无效选择")
    
    def process_manager_menu(self):
        """进程管理菜单"""
        while True:
            print("\n--- 进程管理 ---")
            print("1. 列出进程")
            print("2. 查找进程")
            print("3. 终止进程")
            print("0. 返回主菜单")
            
            choice = input("请选择: ").strip()
            
            if choice == '1':
                processes = self.process_manager.list_processes()
                print(f"找到 {len(processes)} 个进程")
                for proc in processes[:10]:  # 只显示前10个
                    print(f"  PID: {proc['pid']}, 名称: {proc['name']}, "
                          f"CPU: {proc['cpu_percent']:.1f}%, 内存: {proc['memory_percent']:.1f}%")
                if len(processes) > 10:
                    print(f"  ... 还有 {len(processes) - 10} 个进程")
            
            elif choice == '2':
                name = input("请输入进程名称: ").strip()
                processes = self.process_manager.list_processes()
                found = [p for p in processes if name.lower() in p['name'].lower()]
                print(f"找到 {len(found)} 个匹配的进程")
                for proc in found:
                    print(f"  PID: {proc['pid']}, 名称: {proc['name']}")
            
            elif choice == '3':
                try:
                    pid = int(input("请输入进程PID: ").strip())
                    if self.process_manager.kill_process(pid):
                        print(f"进程 {pid} 已终止")
                    else:
                        print(f"终止进程 {pid} 失败")
                except ValueError:
                    print("无效的PID")
            
            elif choice == '0':
                break
            else:
                print("无效选择")
    
    def system_info_menu(self):
        """系统信息菜单"""
        print("\n--- 系统信息 ---")
        print(f"操作系统: {os.name}")
        print(f"平台: {sys.platform}")
        print(f"Python版本: {sys.version}")
        print(f"当前工作目录: {os.getcwd()}")
        print(f"用户: {os.environ.get('USERNAME', os.environ.get('USER', '未知'))}")
        
        # 尝试获取更多系统信息
        try:
            import platform
            print(f"系统: {platform.system()}")
            print(f"节点名称: {platform.node()}")
            print(f"处理器: {platform.processor()}")
        except ImportError:
            pass
    
    def config_menu(self):
        """配置管理菜单"""
        while True:
            print("\n--- 配置管理 ---")
            print("1. 查看配置")
            print("2. 修改监控间隔")
            print("3. 修改告警阈值")
            print("0. 返回主菜单")
            
            choice = input("请选择: ").strip()
            
            if choice == '1':
                print("当前配置:")
                for key, value in self.monitor.config.items():
                    print(f"  {key}: {value}")
            
            elif choice == '2':
                try:
                    interval = int(input("请输入监控间隔(秒): ").strip())
                    self.monitor.config["monitor_interval"] = interval
                    self.monitor.save_config()
                    print(f"监控间隔已设置为 {interval} 秒")
                except ValueError:
                    print("无效的数值")
            
            elif choice == '3':
                try:
                    cpu_threshold = int(input("请输入CPU告警阈值(%): ").strip())
                    mem_threshold = int(input("请输入内存告警阈值(%): ").strip())
                    disk_threshold = int(input("请输入磁盘告警阈值(%): ").strip())
                    
                    self.monitor.config["alert_thresholds"] = {
                        "cpu_usage": cpu_threshold,
                        "memory_usage": mem_threshold,
                        "disk_usage": disk_threshold
                    }
                    self.monitor.save_config()
                    print("告警阈值已更新")
                except ValueError:
                    print("无效的数值")
            
            elif choice == '0':
                break
            else:
                print("无效选择")
    
    def run(self):
        """运行系统管理工具"""
        print("欢迎使用系统管理工具!")
        
        while True:
            self.show_menu()
            choice = input("请选择: ").strip()
            
            if choice == '1':
                self.system_monitor_menu()
            elif choice == '2':
                self.process_manager_menu()
            elif choice == '3':
                self.system_info_menu()
            elif choice == '4':
                self.config_menu()
            elif choice == '0':
                print("感谢使用!")
                break
            else:
                print("无效选择")

# 演示应用
def demo_system_management():
    """演示系统管理应用"""
    print("=== 系统和进程管理演示 ===\n")
    
    # 1. 基本系统信息
    print("1. 基本系统信息:")
    print(f"  操作系统: {os.name}")
    print(f"  平台: {sys.platform}")
    print(f"  Python版本: {sys.version[:50]}...")
    print(f"  当前工作目录: {os.getcwd()}")
    
    # 2. 环境变量
    print(f"\n2. 环境变量:")
    important_vars = ['PATH', 'HOME', 'USER', 'USERNAME']
    for var in important_vars:
        value = os.environ.get(var, '未设置')
        if var == 'PATH':
            print(f"  {var}: {value[:50]}..." if len(value) > 50 else f"  {var}: {value}")
        else:
            print(f"  {var}: {value}")
    
    # 3. 系统监控演示
    print(f"\n3. 系统监控演示:")
    monitor = SystemMonitor()
    info = monitor.get_system_info()
    print(f"  时间: {info['timestamp']}")
    print(f"  CPU使用率: {info['cpu_usage']:.1f}%")
    print(f"  内存使用率: {info['memory_usage']:.1f}%")
    print(f"  磁盘使用率: {info['disk_usage']:.1f}%")
    
    # 4. 进程管理演示
    print(f"\n4. 进程管理演示:")
    process_manager = ProcessManager()
    try:
        processes = process_manager.list_processes()
        print(f"  当前运行进程数: {len(processes)}")
        if processes:
            print("  前5个进程:")
            for proc in processes[:5]:
                print(f"    PID: {proc['pid']}, 名称: {proc['name']}")
    except Exception as e:
        print(f"  进程管理错误: {e}")
    
    # 5. 文件系统操作
    print(f"\n5. 文件系统操作:")
    try:
        import shutil
        total, used, free = shutil.disk_usage('/')
        print(f"  磁盘总空间: {total / (1024**3):.2f} GB")
        print(f"  已使用空间: {used / (1024**3):.2f} GB")
        print(f"  可用空间: {free / (1024**3):.2f} GB")
        print(f"  使用率: {used/total*100:.1f}%")
    except Exception as e:
        print(f"  磁盘信息获取错误: {e}")

# 运行演示
demo_system_management()

# 清理演示文件
config_files = ["monitor_config.json", "system_monitor.log"]
for file in config_files:
    file_path = Path(file)
    if file_path.exists():
        file_path.unlink()
        print(f"清理演示文件: {file}")

最佳实践

import os
import sys
import subprocess
import logging
from pathlib import Path

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class SystemManagementBestPractices:
    """系统管理最佳实践"""
    
    @staticmethod
    def safe_subprocess_execution(command, timeout=30):
        """安全的子进程执行"""
        try:
            # 使用列表形式避免shell注入
            if isinstance(command, str):
                command = command.split()
            
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=timeout,
                check=True  # 非零返回码会抛出异常
            )
            
            logger.info(f"命令执行成功: {' '.join(command)}")
            return result.stdout.strip()
            
        except subprocess.TimeoutExpired:
            logger.error(f"命令执行超时: {' '.join(command)}")
            raise
        except subprocess.CalledProcessError as e:
            logger.error(f"命令执行失败: {' '.join(command)}, 错误: {e.stderr}")
            raise
        except FileNotFoundError:
            logger.error(f"命令未找到: {' '.join(command)}")
            raise
    
    @staticmethod
    def safe_file_operations():
        """安全的文件操作"""
        try:
            # 使用pathlib而不是os.path
            file_path = Path("test_file.txt")
            
            # 安全写入
            with file_path.open('w', encoding='utf-8') as f:
                f.write("Hello, World!")
            
            # 安全读取
            with file_path.open('r', encoding='utf-8') as f:
                content = f.read()
            
            # 安全删除
            if file_path.exists():
                file_path.unlink()
            
            logger.info("文件操作成功")
            return content
            
        except PermissionError:
            logger.error("文件权限不足")
            raise
        except Exception as e:
            logger.error(f"文件操作错误: {e}")
            raise
    
    @staticmethod
    def environment_variable_handling():
        """环境变量处理"""
        # 安全获取环境变量
        database_url = os.environ.get('DATABASE_URL', 'sqlite:///default.db')
        debug_mode = os.environ.get('DEBUG', 'False').lower() in ('true', '1', 'yes')
        
        # 验证环境变量
        required_vars = ['DATABASE_URL']
        missing_vars = [var for var in required_vars if var not in os.environ]
        if missing_vars:
            logger.warning(f"缺少环境变量: {missing_vars}")
        
        return {
            'database_url': database_url,
            'debug_mode': debug_mode,
            'missing_vars': missing_vars
        }
    
    @staticmethod
    def cross_platform_compatibility():
        """跨平台兼容性"""
        platform_info = {
            'system': sys.platform,
            'is_windows': sys.platform.startswith('win'),
            'is_linux': sys.platform.startswith('linux'),
            'is_mac': sys.platform.startswith('darwin')
        }
        
        # 跨平台路径处理
        config_dir = Path.home() / '.config' / 'myapp'
        if platform_info['is_windows']:
            config_dir = Path(os.environ.get('APPDATA', '')) / 'MyApp'
        
        # 跨平台命令执行
        if platform_info['is_windows']:
            ping_command = ['ping', '-n', '1', '127.0.0.1']
        else:
            ping_command = ['ping', '-c', '1', '127.0.0.1']
        
        return {
            'platform': platform_info,
            'config_dir': config_dir,
            'ping_command': ping_command
        }
    
    @staticmethod
    def resource_management():
        """资源管理"""
        import contextlib
        
        @contextlib.contextmanager
        def managed_temp_directory():
            """管理的临时目录"""
            import tempfile
            temp_dir = tempfile.mkdtemp()
            try:
                yield Path(temp_dir)
            finally:
                import shutil
                shutil.rmtree(temp_dir, ignore_errors=True)
        
        @contextlib.contextmanager
        def managed_process(command):
            """管理的进程"""
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            try:
                yield process
            finally:
                if process.poll() is None:
                    process.terminate()
                    try:
                        process.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        process.kill()
        
        return {
            'temp_directory_manager': managed_temp_directory,
            'process_manager': managed_process
        }

def best_practices_demo():
    """最佳实践演示"""
    print("=== 系统管理最佳实践 ===\n")
    
    # 1. 安全子进程执行
    print("1. 安全子进程执行:")
    try:
        result = SystemManagementBestPractices.safe_subprocess_execution(['echo', 'Hello, World!'])
        print(f"  执行结果: {result}")
    except Exception as e:
        print(f"  执行错误: {e}")
    
    # 2. 安全文件操作
    print("\n2. 安全文件操作:")
    try:
        content = SystemManagementBestPractices.safe_file_operations()
        print(f"  文件内容: {content}")
    except Exception as e:
        print(f"  操作错误: {e}")
    
    # 3. 环境变量处理
    print("\n3. 环境变量处理:")
    env_info = SystemManagementBestPractices.environment_variable_handling()
    print(f"  数据库URL: {env_info['database_url']}")
    print(f"  调试模式: {env_info['debug_mode']}")
    if env_info['missing_vars']:
        print(f"  缺少变量: {env_info['missing_vars']}")
    
    # 4. 跨平台兼容性
    print("\n4. 跨平台兼容性:")
    platform_info = SystemManagementBestPractices.cross_platform_compatibility()
    print(f"  系统: {platform_info['platform']['system']}")
    print(f"  配置目录: {platform_info['config_dir']}")
    print(f"  Ping命令: {' '.join(platform_info['ping_command'])}")
    
    # 5. 资源管理
    print("\n5. 资源管理:")
    resources = SystemManagementBestPractices.resource_management()
    print("  上下文管理器已准备就绪")

best_practices_demo()

总结

本篇教程深入探讨了Python中的系统和进程管理，包括：

  1. os模块：基本操作系统功能和文件系统操作
  2. sys模块：Python解释器交互和系统参数
  3. subprocess模块：子进程创建和管理
  4. 进程管理：多进程和并发执行
  5. 线程管理：多线程和线程同步
  6. 系统信息获取：CPU、内存、磁盘等系统信息
  7. 文件系统操作：高级文件和目录操作
  8. 环境变量管理：环境变量的读取和设置
  9. 实际应用示例：综合系统管理工具
  10. 最佳实践：安全和高效的系统管理方法

掌握这些系统和进程管理技术能够帮助您：

  • 构建强大的系统管理工具
  • 实现自动化运维脚本
  • 监控和管理系统资源
  • 管理复杂的进程和线程
  • 创建跨平台的系统应用
  • 处理系统级的配置和环境

Python的系统管理能力使其成为系统管理员、DevOps工程师和需要与操作系统深度集成的开发者的理想选择。

在下一章中，我们将学习网络编程基础，掌握Python在网络应用开发方面的强大功能。

# python # microsoft # 网络
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

AI 工程实践与架构设计

输入一段精简的话，例如：“深度解析 0-1 全自研的外呼机器人架构。本文对比了传统 IVR 与纯 AI Agent 方案的局限性，提出了‘大模型意图识别 + 受控状态机’的混合驱动架构，解决了高延迟与业务合规性痛点，适用于高频线索清洗业务。”

avatar AI Agent技术社区

11 万 Star 的 Spec Kit，90% 的工程师只用它解决了 10% 的问题

什么 Vibe Coding 的天花板这么低先说清楚问题，再说解决方案。vibe coding 这个词是 Andrej Karpathy 提出的，原意是「顺着感觉写代码，让 AI 处理细节」。在原型验证阶段，这个方式非常有效——你描述个大概，AI 给你一个能跑起来的架子，5 分钟内看到效果。但进入真正的功能开发之后，这个模式会遇到一个内在矛盾：AI Agent 不是搜索引擎，它是「字面意思执行者」

avatar AI Agent技术社区

我花了3个月，把一个终端 AI Agent 搬进了浏览器——踩坑全记录

本文记录了将终端AIAgent迁移至浏览器的完整过程。作者最初基于LangChain开发了功能强大的终端Agent，但为提升易用性决定进行浏览器化改造。技术方案历经三次迭代：从WebSocket后端代理（存在资源消耗和状态同步问题）、WebAssembly方案（功能受限），最终选定TypeScript重写核心+WebWorker的架构，实现异步逻辑与UI解耦。开发中攻克了三大核心难题：WebWor

avatar AI Agent技术社区