🤖 实战：自动化脚本

前面 7 篇全是理论基础，这一篇我们把它串起来——用 Python 解决真实世界的自动化需求。

1. 文件批处理

批量重命名

# /opt/kb/docs/python/batch_rename.py
"""按规则批量重命名文件"""
from pathlib import Path
from datetime import datetime

def batch_rename(directory, prefix="", add_date=False):
    """给目录下所有文件加前缀和日期"""
    target = Path(directory)
    if not target.exists():
        print(f"目录 {directory} 不存在")
        return

    files = [f for f in target.iterdir() if f.is_file()]
    for i, file in enumerate(files, 1):
        name_parts = []
        if prefix:
            name_parts.append(prefix)
        if add_date:
            name_parts.append(datetime.now().strftime("%Y%m%d"))
        name_parts.append(f"{i:03d}")

        new_name = "_".join(name_parts) + file.suffix
        new_path = file.with_name(new_name)
        file.rename(new_path)
        print(f"{file.name} → {new_name}")

    print(f"处理完成：{len(files)} 个文件")

# 使用：batch_rename("photos", "旅行", add_date=True)

按日期归档

# /opt/kb/docs/python/archive_by_date.py
"""按文件修改日期自动归档到子目录"""
import shutil
from pathlib import Path
from datetime import datetime

def archive_by_date(directory):
    target = Path(directory)
    for file in target.iterdir():
        if not file.is_file():
            continue

        mtime = datetime.fromtimestamp(file.stat().st_mtime)
        date_dir = target / mtime.strftime("%Y-%m")
        date_dir.mkdir(exist_ok=True)

        shutil.move(str(file), str(date_dir / file.name))
        print(f"{file.name} → {date_dir.name}/")

# 使用：archive_by_date("downloads")

2. 系统监控

# /opt/kb/docs/python/monitor.py
"""系统资源监控，超阈值报警"""
import psutil  # pip install psutil

THRESHOLDS = {
    "cpu": 80,     # CPU 超过 80% 报警
    "memory": 85,  # 内存超过 85% 报警
    "disk": 90,    # 磁盘超过 90% 报警
}

def check_system():
    """检查系统资源，返回告警列表"""
    alerts = []

    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    if cpu_percent > THRESHOLDS["cpu"]:
        alerts.append(f"⚠ CPU 使用率 {cpu_percent}% (阈值 {THRESHOLDS['cpu']}%)")

    # 内存
    mem = psutil.virtual_memory()
    if mem.percent > THRESHOLDS["memory"]:
        alerts.append(f"⚠ 内存使用率 {mem.percent}% (阈值 {THRESHOLDS['memory']}%)")

    # 磁盘
    for part in psutil.disk_partitions():
        usage = psutil.disk_usage(part.mountpoint)
        if usage.percent > THRESHOLDS["disk"]:
            alerts.append(
                f"⚠ 磁盘 {part.mountpoint} 使用率 {usage.percent}% "
                f"(已用 {usage.used // (1024**3)}GB / "
                f"总量 {usage.total // (1024**3)}GB)"
            )

    return alerts

if __name__ == "__main__":
    alerts = check_system()
    if alerts:
        print("🚨 系统告警：")
        for a in alerts:
            print(f"  {a}")
    else:
        print("✅ 系统正常")

3. 网页数据采集（守规矩地爬）

# /opt/kb/docs/python/scrape_demo.py
"""简单的网页数据采集 — 仅用于学习"""
import requests
from bs4 import BeautifulSoup  # pip install beautifulsoup4

def scrape_quotes():
    """采集 quotes.toscrape.com 的名言（公开练习网站）"""
    url = "https://quotes.toscrape.com/"
    resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"})
    resp.raise_for_status()

    soup = BeautifulSoup(resp.text, "html.parser")
    quotes = []

    for quote_div in soup.select(".quote"):
        text = quote_div.select_one(".text").get_text(strip=True)
        author = quote_div.select_one(".author").get_text(strip=True)
        quotes.append({"text": text, "author": author})

    return quotes

# 使用
# for q in scrape_quotes():
#     print(f"「{q['text']}」—— {q['author']}")

爬虫道德守则

1. 先看 robots.txt（如 https://example.com/robots.txt），遵守网站规则
2. 加 User-Agent 标识自己，别冒充浏览器
3. 控制请求频率，别把别人服务器打挂了
4. 只抓公开数据，不抓需要登录的/个人隐私的
5. 遵守相关法律法规，不确定就别爬

4. 自动化邮件

# /opt/kb/docs/python/send_report.py
"""通过 SMTP 发送邮件报告"""
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime

def send_report(to_email, subject, body, smtp_config):
    """发送邮件"""
    msg = MIMEMultipart()
    msg["From"] = smtp_config["user"]
    msg["To"] = to_email
    msg["Subject"] = subject
    msg.attach(MIMEText(body, "plain", "utf-8"))

    with smtplib.SMTP_SSL(smtp_config["host"], smtp_config.get("port", 465)) as server:
        server.login(smtp_config["user"], smtp_config["password"])
        server.send_message(msg)

    print(f"📧 邮件已发送至 {to_email}")

# 使用（以 QQ 邮箱为例）
# smtp_cfg = {
#     "host": "smtp.qq.com",
#     "port": 465,
#     "user": "你的QQ@qq.com",
#     "password": "授权码（非QQ密码！）"  # QQ邮箱 → 设置 → 账户 → POP3/SMTP 开启获取
# }
# send_report("receiver@qq.com", "日报", "今日正常", smtp_cfg)

QQ 邮箱 SMTP

在 QQ 邮箱设置中开启 POP3/SMTP 服务后会得到一个授权码，用它当密码，不能用 QQ 密码。

5. SQLite 数据库

# /opt/kb/docs/python/sqlite_demo.py
"""SQLite 快速上手 — Python 内置，无需安装"""
import sqlite3

# 连接数据库（文件不存在自动创建）
conn = sqlite3.connect("app.db")
cursor = conn.cursor()

# 建表
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# 插入（用参数化查询防 SQL 注入）
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("小雅", "xiaoya@example.com"))
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("小明", "xiaoming@example.com"))
conn.commit()

# 查询
cursor.execute("SELECT * FROM users")
for row in cursor.fetchall():
    print(row)  # (1, '小雅', 'xiaoya@example.com', '...')

# 条件查询
cursor.execute("SELECT name, email FROM users WHERE name LIKE ?", ("小%",))
print(cursor.fetchall())  # [('小雅', ...), ('小明', ...)]

# 更新
cursor.execute("UPDATE users SET email = ? WHERE name = ?", ("new_email@qq.com", "小雅"))
conn.commit()

# 关闭
conn.close()

防 SQL 注入

永远用 ? 占位符传参数，别用字符串拼接 SQL。f"SELECT * FROM users WHERE name = '{user_input}'" 是致命的。

实战：服务器健康日报

# /opt/kb/docs/python/health_report.py
"""服务器健康日报 — 采集 → 格式化 → 输出"""
import os
import sys
import json
import datetime
from collections import OrderedDict

# pip install psutil
try:
    import psutil
    HAS_PSUTIL = True
except ImportError:
    HAS_PSUTIL = False

def collect_health_data():
    """采集服务器健康数据"""
    data = OrderedDict()
    data["报告时间"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    data["主机名"] = os.uname().nodename if hasattr(os, "uname") else os.environ.get("COMPUTERNAME", "未知")

    if HAS_PSUTIL:
        data["CPU使用率"] = f"{psutil.cpu_percent(interval=1)}%"
        mem = psutil.virtual_memory()
        data["内存"] = f"{mem.percent}%（已用 {mem.used//(1024**2)}MB / 总量 {mem.total//(1024**2)}MB）"
        disk = psutil.disk_usage("/")
        data["磁盘"] = f"{disk.percent}%（已用 {disk.used//(1024**3)}GB / 总量 {disk.total//(1024**3)}GB）"
        data["运行时间"] = str(datetime.timedelta(seconds=int(datetime.datetime.now().timestamp() - psutil.boot_time())))
    else:
        data["提示"] = "安装 psutil 获取详细系统信息：pip install psutil"

    return data

def format_report(data, fmt="text"):
    """格式化报告"""
    if fmt == "json":
        return json.dumps(data, indent=2, ensure_ascii=False)

    lines = ["=" * 50, "        📊 服务器健康日报", "=" * 50]
    for k, v in data.items():
        lines.append(f"  {k}: {v}")
    lines.append("=" * 50)
    return "\n".join(lines)

def save_report(report, path="health_report.txt"):
    """保存日报到文件"""
    with open(path, "a", encoding="utf-8") as f:
        f.write(report + "\n\n")
    print(f"📄 报告已保存到 {path}")

if __name__ == "__main__":
    data = collect_health_data()
    report = format_report(data)
    print(report)
    save_report(report)

练习题

1. 写脚本扫描下载目录，把 .jpg、.png 移到 图片/，.pdf、.docx 移到 文档/
2. 用 psutil 写一个后台进程，CPU 超过 90% 时记录日志
3. 爬取一个公开网站（如天气），提取核心信息
4. 写脚本每天自动备份指定目录，文件名为 backup_YYYY-MM-DD.zip
5. 实现完整的健康日报脚本并配置 cron（Linux）或任务计划程序（Windows）定时运行

---

🤖 自动化脚本让你的 Python 技能真正有产出。最后看 常见问题 FAQ，查漏补缺。