欢迎来到 Python 自动化运维专题！学到这里你已经掌握了变量、循环、函数、类、爬虫、数据分析、异步编程等核心技能。现在我们要换个口味——让 Python 成为你的"万能瑞士军刀"，帮你搞定电脑上那些繁琐、重复的日常任务。
这一期不讲究什么高大上的架构，我们就聚焦一个问题：**怎么用 Python 写脚本自动干实事？**
---
## 一、文件系统操作：让文件自己管自己
### 1.1 遍历目录与批量重命名
假设你在下载了一个月的电影资源，名字乱七八糟，你想把它们统一规范一下。用 `os` 和 `pathlib` 就能轻松搞定。
```python
from pathlib import Path
import shutil
from datetime import datetime
defbatch_rename_movie(folder: str) -> None:
"""把文件夹里所有 mp4 文件重命名为 '电影001.mp4' 这种格式"""
dir_path = Path(folder)
files = sorted(dir_path.glob("*.mp4")) # 拿到所有 mp4 文件并排序
for idx, file inenumerate(files, start=1):
new_name = f"电影{idx:03d}{file.suffix}"
new_path = file.with_name(new_name)
if file != new_path:
file.rename(new_path)
print(f"{file.name} → {new_name}")
# 使用示例：
# batch_rename_movie(r"D:\\Downloads\\movies")
```
### 1.2 智能分类整理下载文件夹
更实用的是，根据文件类型自动归类到一个文件夹里的各个子文件夹中。
```python
from pathlib import Path
FOLDER_MAP = {
".jpg": "图片", ".jpeg": "图片", ".png": "图片", ".gif": "图片", ".webp": "图片",
".mp4": "视频", ".avi": "视频", ".mkv": "视频", ".mov": "视频",
".pdf": "文档", ".docx": "文档", ".txt": "文档", ".xlsx": "文档",
".mp3": "音乐", ".flac": "音乐", ".wav": "音乐",
".zip": "压缩包", ".rar": "压缩包", ".7z": "压缩包", ".tar.gz": "压缩包",
".exe": "安装包", ".dmg": "安装包",
}
defsmart_sort(download_dir: str) -> None:
"""把下载目录里的文件分类到对应子文件夹"""
root = Path(download_dir)
for file in root.iterdir():
ifnot file.is_file():
continue
ext = file.suffix.lower()
category = FOLDER_MAP.get(ext, "其他")
dest = root / category / file.name
dest.parent.mkdir(exist_ok=True)
ifnot dest.exists():
shutil.move(str(file), str(dest))
print(f"[{category}] {file.name}")
else:
print(f"[跳过] {file.name} 已存在")
# 使用示例：
# smart_sort(r"D:\\Downloads")
```
运行一次，你的下载文件夹立刻变清爽。
### 1.3 练习
1. 写一个函数，找出某个文件夹下超过 1GB 的大文件，并列出它们的完整路径和大小（以 MB 为单位）。
2. 编写脚本，将指定目录下所有 `.py` 文件的注释行（以 `#` 开头的行）统计出来，输出每份文件的注释行数。
3. 实现一个"重复文件查找器"——扫描一个目录树，打印出 MD5 相同的文件组。提示：使用 `hashlib.md5()` 计算文件摘要。
---
## 二、命令行工具：让终端成为你的第二工作台
### 2.1 用 argparse 编写专业命令行的脚本
Python 内置的 `argparse` 模块让你可以写出像 `ls`、`git` 那样有 `-h` 帮助的命令行程序。
```python
import argparse
import os
from pathlib import Path
defmain():
parser = argparse.ArgumentParser(
prog="fsearch",
description="在指定目录及其子目录中搜索文件名匹配模式的文件",
)
parser.add_argument("pattern", help="搜索模式（支持 * 通配符）")
parser.add_argument("-d", "--directory", default=".", help="搜索根目录（默认当前目录）")
parser.add_argument("-e", "--extension", default=None, help="仅匹配扩展名，如 .py / .txt")
parser.add_argument("--size-gt", type=int, default=None,
help="只显示大于指定字节数的文件")
parser.add_argument("--count-only", action="store_true",
help="只输出结果数量，不显示详细信息")
args = parser.parse_args()
root = Path(args.directory)
ifnot root.is_dir():
print(f"错误：{args.directory} 不是一个有效的目录", flush=True)
return
matches = []
for file in root.rglob(args.pattern):
ifnot file.is_file():
continue
if args.extension and file.suffix != args.extension:
continue
if args.size_gt and file.stat().st_size <= args.size_gt:
continue
matches.append(file)
if args.count_only:
print(f"共找到 {len(matches)} 个匹配文件", flush=True)
else:
for m in matches:
size_mb = m.stat().st_size / (1024 * 1024)
print(f"{m} ({size_mb:.2f} MB)")
print(f"\n总计 {len(matches)} 个文件", flush=True)
if__name__ == "__main__":
main()
```
你可以这样使用：
```bash
# 搜索所有 txt 文件
pythonfsearch"*.txt"-dD:/Documents
# 只看超过 1MB 的 Python 文件
pythonfsearch"*.py"-dC:/Projects--size-gt1048576
# 只返回数量
pythonfsearch"*report*"--count-only
```
### 2.2 subprocess：在 Python 里调用系统命令
有时候光靠 Python 标准库不够用，我们需要调用系统自带的工具（比如 `ping`、`du`、`top`）。
```python
import subprocess
# ===== 用法 1：简单执行 =====
result = subprocess.run(["echo", "Hello from Python"],
capture_output=True, text=True)
print("输出:", result.stdout.strip())
# ===== 用法 2：调用 ping 检查网络 =====
defcheck_host(host: str, timeout: int = 5) -> bool:
"""检查目标主机是否在线"""
param = "-n"if os.name == "nt"else"-c"
cmd = ["ping", param, "1", "-W", str(timeout), host]
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout+2)
return proc.returncode == 0
except subprocess.TimeoutExpired:
returnFalse
# print(check_host("127.0.0.1")) # True
# ===== 用法 3：读取 df/du 命令的输出做磁盘分析 =====
import os
defdisk_usage_summary(top_dir: str = "/") -> dict:
"""获取磁盘使用概要"""
result = subprocess.run(["du", "-sh", top_dir],
capture_output=True, text=True)
lines = result.stdout.strip().split("\n")
info = {}
for line in lines:
parts = line.split()
iflen(parts) >= 2:
info[parts[1]] = parts[0]
return info
# print(disk_usage_summary("."))
```
### 2.3 练习
4. 用 `subprocess` 写一个脚本，监控指定进程（如 `chrome.exe`）的 CPU 和内存占用，每隔 5 秒输出一行报告。
5. 编写一个"定时清理脚本"——每天凌晨 2 点用 `at` 或 `schtasks`（Windows）执行一次，删除临时目录中超过 30 天的文件。
---
## 三、计划任务与邮件通知
### 3.1 用 sched 模块实现轻量级定时任务
不需要装 Celery 或者 cron，标准库就有 `sched` 可以用。
```python
import sched
import time
import os
from pathlib import Path
scheduler = sched.scheduler(time.time, time.sleep)
defclean_temp_files(sc: sched.scheduler):
"""删除 temp 目录下超过 7 天的文件"""
temp_dir = Path.home() / "AppData" / "Local" / "Temp"
cutoff = time.time() - 7 * 86400# 7 天前
cleaned = 0
if temp_dir.exists():
for f in temp_dir.glob("*"):
if f.is_file() and f.stat().st_mtime < cutoff:
f.unlink()
cleaned += 1
print(f"[{time.strftime('%Y-%m-%d %H:%M')}] 已清理 {cleaned} 个过期临时文件")
# 继续安排下一次执行（每天 3 点）
sc.enter(86400, 1, clean_temp_files, (sc,))
# 启动任务
scheduler.enter(0, 1, clean_temp_files, (scheduler,))
try:
scheduler.run()
exceptKeyboardInterrupt:
print("\n任务已停止")
```
### 3.2 发送通知邮件
脚本跑完了，怎么告诉你结果？发邮件最方便。
```python
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
defsend_email(
smtp_server: str,
smtp_port: int,
sender: str,
password: str,
recipient: str,
subject: str,
body: str,
html: bool = False,
) -> bool:
"""通过 SMTP 发送邮件（支持 SSL）"""
msg = MIMEMultipart()
msg["From"] = sender
msg["To"] = recipient
msg["Subject"] = subject
content_type = "html"if html else"plain"
msg.attach(MIMEText(body, content_type, "utf-8"))
try:
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
server.login(sender, password)
server.sendmail(sender, recipient, msg.as_string())
print(f"✅ 邮件已发送至 {recipient}")
returnTrue
exceptExceptionas e:
print(f"❌ 邮件发送失败: {e}")
returnFalse
# 使用示例（以 QQ 邮箱为例）：
# send_email(
# smtp_server="smtp.qq.com",
# smtp_port=465,
# sender="your@qq.com",
# password="your_smtp_authorization_code", # QQ邮箱需要在设置里获取"授权码"
# recipient="admin@company.com",
# subject="服务器监控报告",
# body="CPU 正常，内存 72%，磁盘 45%",
# )
```
### 3.3 练习
6. 将上面的 `clean_temp_files` 和 `send_email` 组合起来：每天清理完临时文件后，自动给你发一封邮件报告清理了多少文件、剩余多少空间。
7. 编写一个"服务器健康检查"脚本：每隔 10 分钟用 `ping` 检查三个站点（127.0.0.1、8.8.8.8、114.114.114.114），如果任何目标不可达就发邮件告警。
---
## 四、注册表与系统信息：Windows 特供
### 4.1 读取系统配置
在 Windows 上我们可以直接读写注册表。下面用一个简单的例子来读取系统信息。
```python
import winreg
import platform
import socket
defget_windows_info() -> dict:
"""收集基础 Windows 系统信息"""
info = {
"hostname": socket.gethostname(),
"platform": platform.platform(),
"python_version": platform.python_version(),
"cpu_count": platform.cpu_count(),
"username": os.environ.get("USERNAME", "unknown"),
}
# 从注册表读取安装日期
try:
key = winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE,
r"SOFTWARE\Microsoft\Windows NT\CurrentVersion",
)
install_date, _ = winreg.QueryValueEx(key, "InstallDate")
info["install_date_timestamp"] = int(install_date)
winreg.CloseKey(key)
exceptFileNotFoundError:
info["install_date"] = "无法读取"
exceptExceptionas e:
info["install_date_error"] = str(e)
return info
for k, v in get_windows_info().items():
print(f" {k}: {v}")
```
>**说明**：`winreg` 只在 Windows 平台可用。如果你在 macOS/Linux 上，可以改用 `psutil` 库来获取跨平台的系统信息。
### 4.2 用 psutil 做系统监控（第三方库）
```bash
pipinstallpsutil
```
```python
import psutil
import time
defsystem_monitor(interval: int = 5, rounds: int = 3) -> None:
"""周期性输出系统资源使用情况"""
print(f"{'时间':>19}{'CPU%':>7}{'内存%':>7}{'磁盘%':>7}")
print("-" * 48)
for _ inrange(rounds):
ts = time.strftime("%Y-%m-%d %H:%M:%S")
cpu = psutil.cpu_percent(interval=None)
mem = psutil.virtual_memory().percent
disk = psutil.disk_usage("/").percent
print(f"{ts:>19}{cpu:>6.1f}% {mem:>6.1f}% {disk:>6.1f}%")
time.sleep(interval)
# system_monitor(interval=5, rounds=5)
```
运行效果大致如下：
```
时间 CPU% 内存% 磁盘%
------------------------------------------------
2026-07-14 09:50:10 12.5% 64.3% 45.2%
2026-07-14 09:50:15 15.8% 64.5% 45.2%
2026-07-14 09:50:20 8.3% 64.4% 45.2%
2026-07-14 09:50:25 22.1% 64.6% 45.2%
2026-07-14 09:50:30 6.7% 64.5% 45.2%
```
---
## 五、综合实战：一键搭建"个人运维仪表盘"
现在我们把前面学的所有东西串在一起，写一个完整的脚本，它可以：
1. 扫描指定目录下的所有文件，统计容量分布
2. 检查系统资源使用情况
3. 把结果保存到一个汇总报告文件里
4. （可选）通过邮件发送给你自己
```python
#!/usr/bin/env python3
"""
daily_health.py — 每日系统健康检查脚本
用法：python daily_health.py [--email recipient@example.com]
"""
import argparse
import json
import os
import platform
import shutil
import subprocess
import time
from datetime import datetime
from pathlib import Path
try:
import psutil
HAS_PSUTIL = True
exceptImportError:
HAS_PSUTIL = False
defscan_directories(dirs: list[str]) -> dict:
"""扫描指定目录的大小分布"""
results = {}
for d in dirs:
p = Path(d)
ifnot p.exists():
results[d] = {"error": "目录不存在"}
continue
total = 0
by_ext: dict[str, int] = {}
file_count = 0
for f in p.rglob("*"):
if f.is_file():
size = f.stat().st_size
total += size
file_count += 1
ext = f.suffix.lower() or"(无扩展名)"
by_ext[ext] = by_ext.get(ext, 0) + size
results[d] = {
"total_bytes": total,
"file_count": file_count,
"by_extension": dict(sorted(by_ext.items(), key=lambdax: -x[1]))[:10],
}
return results
defget_system_info() -> dict:
"""获取系统资源状态"""
info = {"host": platform.node(), "platform": platform.platform()}
ifHAS_PSUTIL:
info.update({
"cpu_percent": psutil.cpu_percent(interval=1),
"memory_total_gb": round(psutil.virtual_memory().total / (1024 ** 3), 1),
"memory_used_percent": psutil.virtual_memory().percent,
"disk_usage_percent": psutil.disk_usage("/").percent,
})
else:
# 回退方案：调用系统命令
try:
out = subprocess.check_output(
["systeminfo", " | findstr /C:\"总物理内存\""],
text=True, shell=True,
)
info["systeminfo_stdout"] = out.strip()
exceptException:
info["psutil_not_available"] = "请安装 pip install psutil"
return info
defbuild_report(scan_dirs: list[str], email: str | None = None) -> str:
"""组装并输出报告"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
scan_data = scan_directories(scan_dirs)
sys_data = get_system_info()
report_lines = [
f"===== 每日系统健康报告 {now} =====",
"",
"--- 系统状态 ---",
json.dumps(sys_data, ensure_ascii=False, indent=2),
"",
"--- 目录扫描 ---",
]
for d, data in scan_data.items():
mb = data.get("total_bytes", 0) / (1024 * 1024)
report_lines.append(f" {d}: {mb:.1f} MB, {data.get('file_count', '?')} 个文件")
for ext, size in data.get("by_extension", {}).items():
kb = size / 1024
if kb > 1024:
report_lines.append(f" .{ext.lstrip('.')}: {kb/1024:.1f} MB")
else:
report_lines.append(f" .{ext.lstrip('.')}: {kb:.0f} KB")
report_text = "\n".join(report_lines) + "\n"
# 保存到文件
out_dir = Path.home() / "AppData" / "Local" / "health_reports"
out_dir.mkdir(parents=True, exist_ok=True)
report_file = out_dir / f"health_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
report_file.write_text(report_text, encoding="utf-8")
print(f"📄 报告已保存到: {report_file}")
# 如果有 psutil 且数据异常，可以触发邮件告警逻辑
# send_email(...) # 参见前面 send_email 函数定义
return report_text
if__name__ == "__main__":
parser = argparse.ArgumentParser(description="每日系统健康检查")
parser.add_argument(
"dirs", nargs="+", help="要扫描的目录列表",
)
parser.add_argument("--email", help="接收报告的邮箱地址（可选）")
args = parser.parse_args()
report = build_report(args.dirs, args.email)
print("\n" + report)
```
运行：
```bash
pythondaily_health.pyD:\DownloadsD:\DocumentsD:\Projects
pythondaily_health.pyD:\ --emailyour@email.com
```
---
## 六、总结与进阶路线
这一期我们学会了：
| 模块 | 用途 | 一句话概括 |
|------|------|-----------|
| `pathlib` / `os` | 文件系统管理 | 像文件管理器一样操控文件夹 |
| `argparse` | 命令行接口 | 让你的脚本像专业工具一样好用 |
| `subprocess` | 调用外部命令 | Python 的"外挂"工具箱 |
| `sched` | 轻量调度器 | 不需要 cron 的定时方案 |
| `smtplib` | 邮件通知 | 脚本执行完，结果到手了 |
| `psutil` | 系统监控 | 实时看 CPU、内存、磁盘 |
| `winreg` | Windows 注册表 | 读取系统安装信息等配置 |
### 下一步怎么走？
自动化运维是通向"工程师思维"的桥梁——不再只是写代码，而是让代码替你工作。接下来的进阶方向：
-**Ansible / SaltStack**：大规模服务器批量运维
-**Django Admin 自定义**：把脚本封装成 Web 管理面板
-**CI/CD 流水线**：用 GitHub Actions 或 Jenkins 自动化测试和部署
---
## 七、练习题参考
以下是前面练习题的参考答案思路：
**练习 1**（查找大文件）：
```python
from pathlib import Path
deffind_large_files(folder: str, threshold_mb: float = 100) -> list[tuple[str, float]]:
results = []
for f in Path(folder).rglob("*"):
if f.is_file() and f.stat().st_size > threshold_mb * 1024 * 1024:
results.append((str(f), f.stat().st_size / (1024 * 1024)))
returnsorted(results, key=lambdax: -x[1])
```
**练习 2**（统计注释行数）：
```python
from pathlib import Path
defcount_comments(folder: str) -> dict[str, int]:
results = {}
for f in Path(folder).rglob("*.py"):
comments = sum(1for line in f.read_text(encoding="utf-8", errors="ignore")
if line.strip().startswith("#"))
results[str(f)] = comments
return results
```
**练习 3**（重复文件检测）：
```python
import hashlib
from pathlib import Path
deffind_duplicates(folder: str) -> dict[str, list[str]]:
defmd5sum(filepath: str) -> str:
h = hashlib.md5()
withopen(filepath, "rb") as f:
h.update(f.read())
return h.hexdigest()
hash_map: dict[str, list[str]] = {}
for f in Path(folder).rglob("*"):
if f.is_file():
h = md5sum(str(f))
hash_map.setdefault(h, []).append(str(f))
# 只保留有多份的文件组
return {k: v for k, v in hash_map.items() iflen(v) > 1}
```
---
**下期预告（Episode 16）**：**Python 与云服务交互**—— 用 `requests` / `boto3` 对接云 API，自动化上传下载 S3/OSS 文件、管理云函数触发器、调用阿里云/腾讯云 OCR 和语音接口，打造真正的"云端自动化"。敬请期待！
10个月宝宝每天需要喝多少奶粉？