今天咱们一起探索Python与操作系统的高级互动技巧。Python不仅是一个强大的编程语言，还能让你轻松操控操作系统，实现自动化任务、文件管理等酷炫功能。让我们从一些基本但实用的命令开始，逐步深入到更复杂的操作。

1. 执行系统命令

Python的subprocess模块是执行外部命令的利器。下面的例子展示了如何运行ls命令并获取其输出：

import subprocess

# 运行ls命令并获取输出

output = subprocess.check_output("ls", shell=True, text=True)

print(output)

小贴士：使用shell=True可以让命令在shell环境下运行。text=True将输出转换为字符串，而非字节串。

2. 重定向标准输出

有时候，你可能想捕获或忽略命令的标准输出。这可以通过subprocess.PIPE实现：

result = subprocess.run(["ls", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

print(result.stdout.decode())

注意：使用subprocess.run替代check_output，可以捕获错误输出。

3. 批量执行命令

当你需要连续执行多个命令时，可以使用分号;分隔它们：

command = "mkdir test; cd test; touch file.txt"

subprocess.run(command, shell=True)

技巧：通过这种方式，你可以构建复杂的脚本逻辑。

4. 文件监控

watchdog库能帮助你监听文件系统变化，非常适合开发实时文件处理应用：

from watchdog.observers import Observer

from watchdog.events import FileSystemEventHandler

class MyHandler(FileSystemEventHandler):

    def on_modified(self, event):

        print(f"Hey, {event.src_path} has been modified!")

if __name__ == "__main__":

    path = "/path/to/watch"

    event_handler = MyHandler()

    observer = Observer()

    observer.schedule(event_handler, path, recursive=True)

    observer.start()

    try:

        while True:

            time.sleep(1)

    except KeyboardInterrupt:

        observer.stop()

    observer.join()

注意事项：确保监听路径正确无误。

5. 系统服务控制

在Linux中，你可以利用systemd服务控制其他程序。下面的脚本展示了如何启动一个服务：

import subprocess

def start_service(service_name):

    subprocess.run(["sudo", "systemctl", "start", service_name])

# 调用函数

start_service("nginx")

安全提醒：使用sudo需要管理员权限，谨慎操作！

6. 系统信息查询

Python的psutil库提供了丰富的系统监控功能，比如查看CPU使用率：

import psutil

# 获取CPU使用率

cpu_percent = psutil.cpu_percent(interval=1)

print(f"CPU Usage: {cpu_percent}%")

技巧：interval参数决定了CPU使用率的计算周期。

7. 文件系统操作

shutil模块提供了高级文件操作，如复制目录：

import shutil

src = "/path/to/source"

dst = "/path/to/destination"

shutil.copytree(src, dst)

注意：确保目标路径不存在，否则会抛出异常。

8. 多线程/多进程

利用concurrent.futures模块，可以轻松实现异步执行系统命令：

import concurrent.futures

def run_command(cmd):

    return subprocess.check_output(cmd, shell=True, text=True)

with concurrent.futures.ThreadPoolExecutor() as executor:

    future_to_cmd = {executor.submit(run_command, cmd): cmd for cmd in ["ls", "pwd"]}

    for future in concurrent.futures.as_completed(future_to_cmd):

        command = future_to_cmd[future]

        try:

            data = future.result()

            print(f"{command} output: {data}")

        except Exception as exc:

            print(f"{command} generated an exception: {exc}")

小贴士：利用ThreadPoolExecutor或ProcessPoolExecutor可以显著提高执行效率。

9. 定时任务

使用schedule库，可以轻松创建定时任务，比如每天凌晨执行清理日志的脚本：

import schedule

import time

def clean_logs():

    # 清理日志的代码

    pass

schedule.every().day.at("00:00").do(clean_logs)

while True:

    schedule.run_pending()

    time.sleep(1)

注意事项：确保你的程序能够持续运行。

10. 网络操作

Python中的socket模块允许你创建网络应用程序。例如，检查服务器端口是否开放：

import socket

def is_port_open(host, port):

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    sock.settimeout(5)

    try:

        sock.connect((host, port))

        print(f"Port {port} is open.")

    except (socket.timeout, ConnectionRefusedError):

        print(f"Port {port} is closed.")

    finally:

        sock.close()

is_port_open('www.example.com', 80)

技巧：设置超时避免无限等待连接。

11. 系统环境变量

你可以使用os.environ来访问和修改环境变量，这对于调试和配置程序非常有用：

import os

# 获取环境变量

print(os.environ['PATH'])

# 设置环境变量

os.environ['MY_VAR'] = 'value'

# 删除环境变量

del os.environ['MY_VAR']

注意事项：修改环境变量可能影响整个系统，谨慎操作。

12. 高级文件操作

使用pathlib模块可以以更直观的方式处理文件路径：

from pathlib import Path

# 创建路径对象

path = Path('/path/to/file')

# 检查文件是否存在

if path.exists():

    print("File exists.")

else:

    print("File does not exist.")

# 创建目录

path.mkdir(parents=True, exist_ok=True)

技巧：parents=True会在不存在的情况下创建父目录。exist_ok=True避免已存在目录的错误。

13. 错误处理与日志记录

在执行系统命令时，错误处理和日志记录至关重要。使用logging模块可以记录详细的执行信息：

import logging

import subprocess

logging.basicConfig(filename='app.log', level=logging.INFO)

def execute_command(cmd):

    try:

        result = subprocess.check_output(cmd, shell=True, text=True)

        logging.info(f"Command executed successfully: {cmd}")

        return result

    except subprocess.CalledProcessError as e:

        logging.error(f"Failed to execute command: {cmd}, error: {e}")

        return None

# 调用函数

output = execute_command("ls")

print(output)

小贴士：日志文件应定期清理，避免占用过多磁盘空间。

14. 自动化部署与脚本化

结合上述所有技术，你可以编写复杂的自动化部署脚本，比如更新网站代码：

import subprocess

import os

def deploy_website():

    # 拉取最新代码

    subprocess.run(["git", "pull"])  

    # 安装依赖

    subprocess.run(["pip", "install", "-r", "requirements.txt"]) 

    # 重启服务

    subprocess.run(["sudo", "service", "nginx", "restart"])

# 调用部署函数

deploy_website()

注意事项：在生产环境中，考虑使用更专业的部署工具，如Docker和Kubernetes。

综合实战案例：自动化日志分析与报警

想象一下，你负责监控一个大型网站的日志文件，需要实时检测异常行为并发送警报。我们可以结合文件监控、系统信息查询和网络操作来实现这一需求：

import psutil

import smtplib

from email.mime.text import MIMEText

from watchdog.observers import Observer

from watchdog.events import FileSystemEventHandler

class LogWatcher(FileSystemEventHandler):

    def on_modified(self, event):

        if "error" in open(event.src_path).read():

            send_alert()

def send_alert():

    msg = MIMEText("Critical error detected in logs!")

    msg['Subject'] = 'Log Alert'

    msg['From'] = 'admin@example.com'

    msg['To'] = 'you@example.com'

    s = smtplib.SMTP('localhost')

    s.send_message(msg)

    s.quit()

if __name__ == "__main__":

    path = "/var/log/nginx/access.log"

    event_handler = LogWatcher()

    observer = Observer()

    observer.schedule(event_handler, path, recursive=False)

    observer.start()

    try:

        while True:

            time.sleep(1)

    except KeyboardInterrupt:

        observer.stop()

    observer.join()

技巧：使用open()读取文件内容，检查特定关键词。考虑增加更复杂的日志解析规则。

注意事项：确保SMTP服务器设置正确。测试报警功能，避免误报。

至此，我们已经探索了Python与操作系统交互的众多高级命令和技巧。

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。