第 4 章:流程自动化

Python 被誉为"胶水语言",在自动化领域有着无可比拟的优势。本章将学习如何用 Python 自动化处理文件、定时任务、Office 文档、邮件和桌面操作,把重复性工作交给代码完成。

4.1 文件自动化处理

文件操作是自动化的基础。Python 提供了现代化的路径操作库 pathlib,相比传统的 os.path 更加面向对象、跨平台。

ℹ️与 JS 的对比

Python 的 pathlib.Path 类似 Node.js 的 path 模块 + fs.promises 的组合,但 API 更加统一和面向对象。

pathlib:现代路径操作

python
"""pathlib 示例 - 现代化的文件路径操作.

对比 Node.js:
  Node: const path = require('path'); path.join(__dirname, 'src', 'index.js')
  Python: Path(__file__).parent / 'src' / 'index.py'
"""

from pathlib import Path

# 创建路径对象(自动处理 Windows/Unix 路径分隔符)
project_root = Path.home() / "projects" / "my-app"
# Windows: C:\Users\xxx\projects\my-app
# Linux/macOS: /home/xxx/projects/my-app

# 路径拼接使用 / 运算符(比 os.path.join 更直观)
config_file = project_root / "config" / "settings.json"

# 常用属性
print(config_file.name)       # settings.json(文件名)
print(config_file.stem)       # settings(不含扩展名)
print(config_file.suffix)     # .json(扩展名)
print(config_file.parent)     # .../my-app/config(父目录)
print(config_file.parts)      # ('...', 'my-app', 'config', 'settings.json')

# 检查存在性
if config_file.exists():
    print(f"文件存在,大小: {config_file.stat().st_size} bytes")

# 创建目录(类似 mkdir -p)
data_dir = project_root / "data" / "backup"
data_dir.mkdir(parents=True, exist_ok=True)

# 读取文本文件(自动处理编码)
content = config_file.read_text(encoding="utf-8")

# 写入文本文件
output_file = project_root / "output.txt"
output_file.write_text("Hello, Python!\n", encoding="utf-8")

# 遍历目录(类似 fs.readdirSync)
for py_file in project_root.glob("**/*.py"):
    print(py_file)

# 重命名
old_file = project_root / "old_name.txt"
new_file = project_root / "new_name.txt"
old_file.rename(new_file)

# 删除文件
if new_file.exists():
    new_file.unlink()

# 删除空目录
data_dir.rmdir()  # 只能删除空目录

shutil:文件复制与移动

shutil 是 Python 标准库中的高级文件操作模块,提供复制、移动、压缩等功能。

python
import shutil
from pathlib import Path

src = Path("source.txt")
dst = Path("backup.txt")

# 复制文件(类似 cp)
shutil.copy2(src, dst)  # copy2 保留元数据(修改时间等)

# 复制目录树(类似 cp -r)
shutil.copytree("src_dir", "dst_dir", dirs_exist_ok=True)

# 移动文件/目录(类似 mv)
shutil.move("old_location", "new_location")

# 删除目录树(类似 rm -rf,慎用!)
shutil.rmtree("temp_dir")

# 创建压缩包
shutil.make_archive("backup", "zip", "data_dir")
# 生成 backup.zip

# 解压
shutil.unpack_archive("backup.zip", "extract_dir")

glob:文件模式匹配

glob 模块支持 shell 风格的通配符匹配,在批量处理文件时非常实用。

python
import glob
from pathlib import Path

# 匹配当前目录所有 .py 文件(类似 ls *.py)
py_files = glob.glob("*.py")

# 递归匹配(类似 find . -name "*.py")
all_py = glob.glob("**/*.py", recursive=True)

# 使用 pathlib 的 glob(推荐)
project = Path(".")

# 所有 Python 文件
for f in project.rglob("*.py"):  # rglob = recursive glob
    print(f)

# 匹配特定模式
for f in project.glob("test_*.py"):
    print(f"测试文件: {f}")

# 排除某些文件
py_files = [f for f in project.rglob("*.py") if "venv" not in str(f)]

watchdog:文件监控

watchdog 库可以监控文件系统事件,实现"文件变化时自动执行某操作"的功能,类似前端开发中的热重载(HMR)。

python
"""文件监控示例 - 类似前端 webpack 的 watch 模式.

安装: pip install watchdog
"""

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class MyHandler(FileSystemEventHandler):
    """自定义事件处理器."""

    def on_modified(self, event):
        """文件被修改时触发."""
        if not event.is_directory:
            print(f"文件被修改: {event.src_path}")
            # 在这里执行自动化操作,比如重新生成报告
            self.process_file(event.src_path)

    def on_created(self, event):
        """新文件创建时触发."""
        print(f"新文件: {event.src_path}")

    def process_file(self, filepath):
        """处理文件的逻辑."""
        print(f"正在处理: {filepath}")


# 监控当前目录
observer = Observer()
observer.schedule(MyHandler(), path=".", recursive=True)
observer.start()

print("开始监控文件变化...按 Ctrl+C 停止")
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()

observer.join()

4.2 定时任务

自动化离不开定时执行。从简单的脚本调度到复杂的任务编排,Python 有多种方案可选。

schedule:简单定时库

schedule 是一个轻量级定时任务库,API 设计非常直观,适合简单的定时需求。

python
"""schedule 示例 - 类似 JS 的 node-schedule 但更简洁.

安装: pip install schedule
"""

import schedule
import time


def job():
    """定时执行的任务."""
    print("执行任务...")


def backup_database():
    """备份数据库."""
    print("正在备份数据库...")


# 每 10 分钟执行一次
schedule.every(10).minutes.do(job)

# 每小时执行
schedule.every().hour.do(job)

# 每天特定时间
schedule.every().day.at("10:30").do(job)

# 工作日执行
schedule.every().monday.do(backup_database)
schedule.every().wednesday.at("13:15").do(job)

# 更复杂的间隔
schedule.every(5).to(10).minutes.do(job)  # 每 5-10 分钟随机执行

# 主循环
while True:
    schedule.run_pending()
    time.sleep(1)

APScheduler:高级调度框架

APScheduler(Advanced Python Scheduler)是 Python 最成熟的调度框架,支持多种触发器和执行器,功能类似 Node.js 的 node-cron + bull 队列的组合。

python
"""APScheduler 示例 - 企业级定时任务框架.

安装: pip install apscheduler
"""

from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger


def send_report():
    """发送日报."""
    print(f"[{datetime.now()}] 发送日报")


def clean_temp_files():
    """清理临时文件."""
    print(f"[{datetime.now()}] 清理临时文件")


# 创建调度器(后台运行,不阻塞主线程)
scheduler = BackgroundScheduler()

# 1. 间隔触发(每 30 分钟)
scheduler.add_job(
    send_report,
    trigger=IntervalTrigger(minutes=30),
    id="send_report",
    name="发送日报"
)

# 2. Cron 触发(类似 Linux crontab)
# 每天上午 9 点执行
scheduler.add_job(
    send_report,
    trigger=CronTrigger(hour=9, minute=0),
    id="morning_report",
    name="晨报"
)

# 每周一早上 8 点执行
scheduler.add_job(
    clean_temp_files,
    trigger=CronTrigger(day_of_week="mon", hour=8, minute=0),
    id="weekly_cleanup",
    name="每周清理"
)

# 3. 日期触发(特定时间执行一次)
from apscheduler.triggers.date import DateTrigger
scheduler.add_job(
    send_report,
    trigger=DateTrigger(run_date=datetime(2024, 12, 31, 23, 59)),
    id="year_end"
)

# 启动调度器
scheduler.start()

print("调度器已启动,按 Ctrl+C 退出")
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    scheduler.shutdown()
    print("调度器已关闭")

系统级定时任务

对于生产环境,通常使用系统级定时任务,而不是在 Python 进程中内嵌调度器。

bash
# Linux/macOS: crontab
# 编辑定时任务
crontab -e

# 格式: 分 时 日 月 周 命令
# 每天凌晨 2 点执行 Python 脚本
0 2 * * * /usr/bin/python3 /home/user/scripts/backup.py

# 每 5 分钟执行
*/5 * * * * /usr/bin/python3 /home/user/scripts/monitor.py

# 工作日每小时执行
0 * * * 1-5 /usr/bin/python3 /home/user/scripts/workday_task.py

# Windows: 任务计划程序(Task Scheduler)
# 或使用 schtasks 命令行
schtasks /create /tn "DailyBackup" /tr "python C:\scripts\backup.py" /sc daily /st 02:00

4.3 Excel 自动化

处理 Excel 是办公自动化的常见需求。openpyxl 是读写 .xlsx 格式的事实标准库。

python
"""openpyxl 示例 - Excel 自动化处理.

安装: pip install openpyxl
"""

from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter


# ========== 创建新工作簿 ==========
wb = Workbook()
ws = wb.active  # 获取活动工作表
ws.title = "销售数据"

# 写入数据(类似操作二维数组)
# 第 1 行表头
headers = ["产品", "销量", "单价", "总额"]
for col, header in enumerate(headers, 1):
    cell = ws.cell(row=1, column=col, value=header)
    # 设置表头样式
    cell.font = Font(bold=True, color="FFFFFF")
    cell.fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
    cell.alignment = Alignment(horizontal="center")

# 写入数据行
data = [
    ["iPhone 15", 100, 5999, 599900],
    ["MacBook Pro", 50, 14999, 749950],
    ["AirPods Pro", 200, 1899, 379800],
]

for row_idx, row_data in enumerate(data, 2):
    for col_idx, value in enumerate(row_data, 1):
        ws.cell(row=row_idx, column=col_idx, value=value)

# 调整列宽
ws.column_dimensions["A"].width = 15
ws.column_dimensions["B"].width = 10
ws.column_dimensions["C"].width = 10
ws.column_dimensions["D"].width = 12

# 添加边框
thin_border = Border(
    left=Side(style="thin"),
    right=Side(style="thin"),
    top=Side(style="thin"),
    bottom=Side(style="thin")
)
for row in ws.iter_rows(min_row=1, max_row=4, min_col=1, max_col=4):
    for cell in row:
        cell.border = thin_border

# 保存
wb.save("sales_report.xlsx")


# ========== 读取现有工作簿 ==========
wb2 = load_workbook("sales_report.xlsx")
ws2 = wb2["销售数据"]

# 读取单元格
print(ws2["A1"].value)  # 产品
print(ws2["B2"].value)  # 100

# 遍历所有行
for row in ws2.iter_rows(min_row=2, values_only=True):
    print(row)  # ('iPhone 15', 100, 5999, 599900)

# 获取最大行数和列数
print(f"行数: {ws2.max_row}, 列数: {ws2.max_column}")


# ========== 生成图表 ==========
from openpyxl.chart import BarChart, Reference

chart = BarChart()
chart.type = "col"
chart.title = "产品销量对比"
chart.y_axis.title = "销量"
chart.x_axis.title = "产品"

data_ref = Reference(ws, min_col=2, min_row=1, max_row=4)
cats_ref = Reference(ws, min_col=1, min_row=2, max_row=4)
chart.add_data(data_ref, titles_from_data=True)
chart.set_categories(cats_ref)

ws.add_chart(chart, "F2")
wb.save("sales_report_with_chart.xlsx")
💡技巧

如果需要处理 .xls(旧版 Excel)格式,使用 xlrd 读取;如果需要高性能处理大数据量 Excel,考虑 pandas.read_excel + pandas.DataFrame.to_excel

4.4 Word/PDF 自动化

自动生成 Word 和 PDF 文档是报告自动化、合同生成的常见需求。

python-docx:生成 Word 文档

python
"""python-docx 示例 - 自动生成 Word 报告.

安装: pip install python-docx
"""

from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH

# 创建文档
doc = Document()

# 添加标题
title = doc.add_heading("月度销售报告", level=0)
title.alignment = WD_ALIGN_PARAGRAPH.CENTER

# 添加段落
p = doc.add_paragraph("本月销售数据汇总如下:")
p.add_run("(数据截止至 2024-12-31)").italic = True

# 添加表格
table = doc.add_table(rows=1, cols=4)
table.style = "Light Grid Accent 1"

# 表头
hdr_cells = table.rows[0].cells
headers = ["产品", "销量", "单价", "总额"]
for i, header in enumerate(headers):
    hdr_cells[i].text = header
    # 加粗表头
    for paragraph in hdr_cells[i].paragraphs:
        for run in paragraph.runs:
            run.font.bold = True

# 数据行
sales_data = [
    ("iPhone 15", "100", "¥5,999", "¥599,900"),
    ("MacBook Pro", "50", "¥14,999", "¥749,950"),
]
for item, qty, price, total in sales_data:
    row = table.add_row().cells
    row[0].text = item
    row[1].text = qty
    row[2].text = price
    row[3].text = total

# 添加分页
doc.add_page_break()

# 添加二级标题
doc.add_heading("详细分析", level=1)
doc.add_paragraph("iPhone 15 本月销量领先,主要得益于...")

# 保存
doc.save("monthly_report.docx")

ReportLab:生成 PDF

python
"""ReportLab 示例 - 生成 PDF 文档.

安装: pip install reportlab
"""

from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import cm
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
from reportlab.lib import colors


def create_pdf(filename: str) -> None:
    """创建 PDF 报告."""
    doc = SimpleDocTemplate(
        filename,
        pagesize=A4,
        rightMargin=2 * cm,
        leftMargin=2 * cm,
        topMargin=2 * cm,
        bottomMargin=2 * cm,
    )

    styles = getSampleStyleSheet()
    story = []

    # 标题
    title_style = ParagraphStyle(
        "CustomTitle",
        parent=styles["Heading1"],
        fontSize=24,
        textColor=colors.HexColor("#1a1a2e"),
        spaceAfter=30,
        alignment=1,  # 居中
    )
    story.append(Paragraph("销售报告", title_style))
    story.append(Spacer(1, 0.5 * cm))

    # 正文
    story.append(Paragraph("本报告汇总了 Q4 季度销售数据。", styles["Normal"]))
    story.append(Spacer(1, 0.5 * cm))

    # 表格数据
    data = [
        ["产品", "销量", "单价", "总额"],
        ["iPhone 15", "100", "¥5,999", "¥599,900"],
        ["MacBook Pro", "50", "¥14,999", "¥749,950"],
        ["合计", "150", "-", "¥1,349,850"],
    ]

    table = Table(data, colWidths=[4 * cm, 3 * cm, 3 * cm, 3 * cm])
    table.setStyle(TableStyle([
        ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#366092")),
        ("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
        ("ALIGN", (0, 0), (-1, -1), "CENTER"),
        ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
        ("FONTSIZE", (0, 0), (-1, 0), 12),
        ("BOTTOMPADDING", (0, 0), (-1, 0), 12),
        ("BACKGROUND", (0, 1), (-1, -2), colors.HexColor("#f0f0f0")),
        ("GRID", (0, 0), (-1, -1), 1, colors.grey),
        ("FONTNAME", (0, -1), (-1, -1), "Helvetica-Bold"),
        ("BACKGROUND", (0, -1), (-1, -1), colors.HexColor("#d9e1f2")),
    ]))

    story.append(table)

    # 生成 PDF
    doc.build(story)
    print(f"PDF 已生成: {filename}")


create_pdf("report.pdf")

4.5 邮件自动化

自动发送邮件是系统通知、日报推送的常见需求。Python 提供了从底层到高层的多种邮件发送方案。

smtplib:标准库方案

python
"""使用 smtplib 发送邮件(标准库).

注意:大多数邮箱需要开启 SMTP 授权码,不能使用登录密码。
"""

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
from pathlib import Path


def send_email(
    subject: str,
    body: str,
    to_email: str,
    from_email: str,
    password: str,
    smtp_server: str = "smtp.qq.com",
    smtp_port: int = 587,
    attachment_path: str | None = None,
) -> None:
    """发送邮件,支持附件.

    参数:
        subject: 邮件主题
        body: 邮件正文(HTML 格式)
        to_email: 收件人地址
        from_email: 发件人地址
        password: SMTP 授权码(不是登录密码)
        smtp_server: SMTP 服务器地址
        smtp_port: SMTP 端口
        attachment_path: 附件路径(可选)
    """
    # 创建多部分邮件
    msg = MIMEMultipart("alternative")
    msg["Subject"] = subject
    msg["From"] = from_email
    msg["To"] = to_email

    # 添加 HTML 正文
    html_part = MIMEText(body, "html", "utf-8")
    msg.attach(html_part)

    # 添加附件
    if attachment_path and Path(attachment_path).exists():
        with open(attachment_path, "rb") as f:
            attachment = MIMEBase("application", "octet-stream")
            attachment.set_payload(f.read())

        encoders.encode_base64(attachment)
        filename = Path(attachment_path).name
        attachment.add_header(
            "Content-Disposition",
            f"attachment; filename= {filename}",
        )
        msg.attach(attachment)

    # 连接 SMTP 服务器并发送
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()  # 启用 TLS 加密
        server.login(from_email, password)
        server.sendmail(from_email, to_email, msg.as_string())

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


# 使用示例
html_body = """

  
    

日报通知

今日数据汇总:

  • 新用户: 150 人
  • 订单量: 320 单
  • 销售额: ¥45,000
""" # send_email( # subject="【日报】2024-12-31 数据汇总", # body=html_body, # to_email="recipient@example.com", # from_email="your_email@qq.com", # password="your_smtp_auth_code", # attachment_path="report.xlsx", # )

yagmail:简化版邮件发送

yagmail 是 smtplib 的高级封装,API 极其简洁,一行代码即可发送邮件。

python
"""yagmail 示例 - 极简邮件发送.

安装: pip install yagmail
"""

import yagmail


def send_simple_email() -> None:
    """使用 yagmail 发送邮件。"""
    # 初始化(只需一次)
    yag = yagmail.SMTP(
        user="your_email@qq.com",
        password="your_auth_code",  # SMTP 授权码
        host="smtp.qq.com",
    )

    # 发送邮件(一行代码!)
    yag.send(
        to="recipient@example.com",
        subject="测试邮件",
        contents="这是邮件正文,支持 \u003cb>HTML",
        attachments="report.pdf",
    )

    print("邮件发送成功")


# 发送给多人
# yag.send(
#     to=["a@example.com", "b@example.com"],
#     cc="boss@example.com",  # 抄送
#     subject="群发测试",
#     contents=["正文", "

HTML 内容

", yagmail.inline("image.png")], # )

4.6 桌面 GUI 自动化

pyautogui 可以控制鼠标和键盘、截取屏幕、识别图像,实现桌面端的 RPA(机器人流程自动化)。

⚠️注意

pyautogui 控制的是真实的鼠标键盘,运行脚本时请确保有安全的停止方式(默认将鼠标移到屏幕左上角会触发异常中断)。不要在生产环境或重要工作时随意运行。

python
"""pyautogui 示例 - 桌面自动化.

安装: pip install pyautogui
"""

import pyautogui
import time

# 安全设置:将鼠标移到屏幕左上角会触发异常中断
pyautogui.FAILSAFE = True

# 获取屏幕尺寸
screen_width, screen_height = pyautogui.size()
print(f"屏幕分辨率: {screen_width}x{screen_height}")

# 获取当前鼠标位置
current_x, current_y = pyautogui.position()
print(f"鼠标位置: ({current_x}, {current_y})")

# 移动鼠标(绝对坐标)
pyautogui.moveTo(100, 200, duration=0.5)  # 0.5 秒内移动到 (100, 200)

# 相对移动
pyautogui.moveRel(50, 0, duration=0.5)  # 向右移动 50 像素

# 点击
pyautogui.click()           # 当前位置单击
pyautogui.click(100, 200)   # 在指定位置单击
pyautogui.doubleClick()     # 双击
pyautogui.rightClick()      # 右键

# 输入文本(类似键盘输入)
pyautogui.typewrite("Hello, World!", interval=0.1)  # 每个字符间隔 0.1 秒

# 按下特殊键
pyautogui.press("enter")
pyautogui.hotkey("ctrl", "c")   # Ctrl+C
pyautogui.hotkey("ctrl", "v")   # Ctrl+V
pyautogui.hotkey("alt", "tab")  # Alt+Tab

# 截图
screenshot = pyautogui.screenshot()
screenshot.save("screenshot.png")

# 区域截图
region = pyautogui.screenshot(region=(0, 0, 300, 400))  # x, y, width, height
region.save("region.png")

# 图像识别(在屏幕上查找图片位置)
# button_location = pyautogui.locateOnScreen("button.png")
# if button_location:
#     center = pyautogui.center(button_location)
#     pyautogui.click(center)

# 等待图片出现(带超时)
# location = pyautogui.locateOnScreen("submit.png", confidence=0.9)

4.7 综合案例:自动整理下载文件夹

结合本章所学,实现一个实用的自动化脚本:按文件类型自动整理下载文件夹。

python
#!/usr/bin/env python3
"""下载文件夹自动整理工具.

功能:
    - 按文件扩展名分类到不同子目录
    - 支持图片、文档、视频、压缩包等常见类型
    - 重复文件自动重命名
    - 空文件夹自动清理

使用方法:
    python organize_downloads.py
    # 或指定目录
    python organize_downloads.py /path/to/downloads
"""

import shutil
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path


# 文件类型分类映射
CATEGORY_MAP = {
    # 图片
    "图片": [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".svg", ".webp", ".ico"],
    # 文档
    "文档": [".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx", ".txt", ".md"],
    # 视频
    "视频": [".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv", ".webm"],
    # 音频
    "音频": [".mp3", ".wav", ".flac", ".aac", ".ogg", ".m4a"],
    # 压缩包
    "压缩包": [".zip", ".rar", ".7z", ".tar", ".gz", ".bz2"],
    # 程序
    "程序": [".exe", ".msi", ".dmg", ".pkg", ".deb", ".rpm"],
    # 代码
    "代码": [".py", ".js", ".ts", ".html", ".css", ".java", ".go", ".rs", ".cpp", ".c"],
}

# 反向查找表:扩展名 -> 分类
EXT_TO_CATEGORY = {}
for category, extensions in CATEGORY_MAP.items():
    for ext in extensions:
        EXT_TO_CATEGORY[ext.lower()] = category


def get_category(file_path: Path) -> str:
    """根据文件扩展名返回分类目录名.

    未知类型归入"其他"。
    """
    ext = file_path.suffix.lower()
    return EXT_TO_CATEGORY.get(ext, "其他")


def get_unique_path(target_dir: Path, filename: str) -> Path:
    """生成不重复的文件路径.

    如果目标文件已存在,自动添加序号后缀。
    类似于操作系统的"副本"命名逻辑。
    """
    target = target_dir / filename
    if not target.exists():
        return target

    # 分离文件名和扩展名
    stem = Path(filename).stem
    suffix = Path(filename).suffix
    counter = 1

    while True:
        new_name = f"{stem}_{counter}{suffix}"
        target = target_dir / new_name
        if not target.exists():
            return target
        counter += 1


def organize_directory(downloads_dir: Path) -> dict[str, int]:
    """整理下载文件夹.

    返回统计信息字典。
    """
    if not downloads_dir.exists():
        print(f"目录不存在: {downloads_dir}")
        return {}

    stats = defaultdict(int)

    # 遍历目录中的文件(不包括子目录)
    for item in downloads_dir.iterdir():
        if not item.is_file():
            continue

        # 跳过本脚本自身
        if item.name == Path(__file__).name:
            continue

        # 确定分类
        category = get_category(item)
        target_dir = downloads_dir / category

        # 创建分类目录
        target_dir.mkdir(exist_ok=True)

        # 处理重复文件名
        target_path = get_unique_path(target_dir, item.name)

        # 移动文件
        shutil.move(str(item), str(target_path))
        print(f"移动: {item.name} -> {category}/{target_path.name}")

        stats[category] += 1

    # 清理空文件夹
    for subdir in downloads_dir.iterdir():
        if subdir.is_dir() and subdir.name != "其他":
            try:
                subdir.rmdir()  # 只能删除空目录
                print(f"删除空目录: {subdir.name}")
            except OSError:
                pass  # 目录不为空,忽略

    return dict(stats)


def main() -> None:
    """主函数."""
    # 获取目标目录
    if len(sys.argv) > 1:
        downloads_dir = Path(sys.argv[1])
    else:
        # 默认使用用户下载目录
        downloads_dir = Path.home() / "Downloads"

    print(f"整理目录: {downloads_dir}")
    print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("-" * 50)

    stats = organize_directory(downloads_dir)

    print("-" * 50)
    print("整理完成!")
    print(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    if stats:
        print("\n统计:")
        total = sum(stats.values())
        for category, count in sorted(stats.items()):
            print(f"  {category}: {count} 个文件")
        print(f"  总计: {total} 个文件")
    else:
        print("没有需要整理的文件。")


if __name__ == "__main__":
    main()
💡扩展思路
  1. 结合 4.2 节的 APScheduler,可以设置每天自动运行整理脚本。
  2. 结合 4.1 节的 watchdog,可以在文件下载完成时立即触发整理。
  3. 添加配置文件(如 JSON/YAML),让用户自定义分类规则。
  4. 添加日志记录,追踪每次整理操作的详细记录。

4.8 subprocess —— 调用外部命令

subprocess 是 Python 标准库中调用外部命令和程序的模块。它等同于在终端中执行 shell 命令, 但可以通过 Python 控制输入输出和错误处理。前端开发者可以把它理解为 Node.js 的 child_process

python
import subprocess
import sys

# 1. 运行命令并获取输出(类似 Node.js 的 execSync)
result = subprocess.run(["git", "status"], capture_output=True, text=True)
print("返回码:", result.returncode)    # 0 表示成功
print("输出:", result.stdout)

# 如果命令失败,抛出异常
result.check_returncode()

# 2. 运行命令并实时输出(类似 pipe stdout)
# subprocess.run(["python", "script.py"], check=True)

# 3. 使用 shell 模式(需要拼接复杂命令时)
# 注意:shell=True 有安全风险,避免用于用户输入
result = subprocess.run(
    "dir" if sys.platform == "win32" else "ls -la",
    shell=True,
    capture_output=True,
    text=True
)

# 4. 管道操作(类似 shell 的 |)
# 示例:echo "hello" | wc -c
echo = subprocess.run(["echo", "hello"], capture_output=True, text=True)
count = subprocess.run(["wc", "-c"], input=echo.stdout, capture_output=True, text=True)
print(f"字符数: {count.stdout.strip()}")

# 5. 运行命令并设置超时
try:
    result = subprocess.run(
        ["sleep", "10"],
        timeout=5,           # 5 秒超时
        capture_output=True
    )
except subprocess.TimeoutExpired:
    print("命令执行超时!")

# 6. 常见实用示例
# 获取当前 Git 分支
result = subprocess.run(
    ["git", "rev-parse", "--abbrev-ref", "HEAD"],
    capture_output=True, text=True
)
branch = result.stdout.strip()
print(f"当前分支: {branch}")

# 安装 Python 包(在自动化脚本中)
# subprocess.run([sys.executable, "-m", "pip", "install", "requests"])
⚠️安全提示

永远不要对用户输入使用 shell=True,这会导致命令注入漏洞。 尽量使用列表参数形式(["git", "status"]),而非字符串形式。

4.9 CLI 参数解析

构建命令行工具是自动化的核心需求。Python 有两条主要路线:标准库的 argparse(零依赖) 和第三方的 click(装饰器风格,更简洁)。

argparse —— 标准库方案

python
"""CLI 示例:文件搜索工具 —— 使用 argparse"""
import argparse
from pathlib import Path

def search_files(directory, pattern, recursive=False, max_results=10):
    """搜索文件"""
    path = Path(directory)
    if not path.exists():
        print(f"目录不存在: {directory}")
        return

    files = path.rglob(pattern) if recursive else path.glob(pattern)
    for i, f in enumerate(files):
        if i >= max_results:
            print(f"...(仅显示前 {max_results} 条)")
            break
        print(f"  {f}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="文件搜索工具 - 按模式查找文件"
    )
    # 位置参数(必填)
    parser.add_argument("directory", help="搜索目录")
    # 可选参数
    parser.add_argument("-p", "--pattern", default="*.py", help="文件模式(默认: *.py)")
    parser.add_argument("-r", "--recursive", action="store_true", help="递归查找")
    parser.add_argument("-n", "--max-results", type=int, default=10, help="最大结果数")

    args = parser.parse_args()
    search_files(args.directory, args.pattern, args.recursive, args.max_results)

# 使用方式:
# python search.py .                    # 搜索当前目录的 .py 文件
# python search.py . -p "*.txt" -r -n 5 # 递归搜索 .txt,最多 5 条

click —— 优雅的第三方方案

python
"""CLI 示例:文件搜索工具 —— 使用 click (pip install click)"""
import click
from pathlib import Path

@click.command()
@click.argument("directory", type=click.Path(exists=True))
@click.option("-p", "--pattern", default="*.py", help="文件模式")
@click.option("-r", "--recursive", is_flag=True, help="递归查找")
@click.option("-n", "--max-results", default=10, help="最大结果数")
def search(directory, pattern, recursive, max_results):
    """按模式搜索文件"""
    path = Path(directory)
    files = path.rglob(pattern) if recursive else path.glob(pattern)
    count = 0
    for f in files:
        if count >= max_results:
            click.echo(f"...(仅显示前 {max_results} 条)")
            break
        click.echo(f"  {f}")
        count += 1

if __name__ == "__main__":
    search()

# click 的优势:
# - 装饰器风格,代码更清晰
# - 自动生成 --help 帮助信息
# - 支持颜色输出、进度条等高级功能
# - 命令组(类似 git 的子命令: git add, git commit)
ℹ️argparse vs click
维度argparseclick
依赖标准库,零依赖需 pip install
风格面向过程构建 parser装饰器声明参数
复杂命令组需要手动实现内置支持
输出美化基本颜色、进度条原生支持
适用场景简单脚本、不需要额外依赖复杂 CLI 工具(如 poetry、black)

4.10 其他实用自动化工具

Pillow —— 图像处理

Pillow 是 Python 最流行的图像处理库,可以批量处理图片:

python
"""Pillow 示例 —— 批量图像处理 (pip install Pillow)"""
from pathlib import Path
from PIL import Image

# 批量缩放图片
def batch_resize(input_dir, output_dir, size=(800, 600)):
    """将目录中所有图片缩放到指定大小"""
    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)

    for img_path in Path(input_dir).glob("*.jpg"):
        with Image.open(img_path) as img:
            img.thumbnail(size)  # 等比缩放,保持宽高比
            img.save(output / img_path.name)
            print(f"处理: {img_path.name} ({img.size})")

# 批量添加水印
def add_watermark(img_path, output_path, text):
    """给图片添加文字水印"""
    from PIL import ImageDraw, ImageFont

    with Image.open(img_path) as img:
        draw = ImageDraw.Draw(img)
        # 在右下角添加水印
        draw.text(
            (img.width - 200, img.height - 40),
            text,
            fill=(255, 255, 255, 128)
        )
        img.save(output_path)

# 批量转换格式
def convert_format(input_dir, to_format="webp"):
    """将目录中所有图片转为指定格式"""
    for img_path in Path(input_dir).glob("*"):
        if img_path.suffix.lower() in [".jpg", ".jpeg", ".png"]:
            img = Image.open(img_path)
            new_path = img_path.with_suffix(f".{to_format}")
            img.save(new_path, to_format.upper())
            print(f"转换: {img_path.name} -> {new_path.name}")

rich —— 终端美化输出

rich 让 Python 的终端输出变得生动美观。它可以美化表格、添加进度条、彩色日志,甚至渲染 Markdown:

python
"""rich 示例 —— 让终端输出更好看 (pip install rich)"""
from rich.console import Console
from rich.table import Table
from rich.progress import track
from rich import print as rprint
import time

console = Console()

# 1. 美化表格
table = Table(title="项目统计")
table.add_column("文件", style="cyan")
table.add_column("行数", style="green", justify="right")
table.add_column("状态", style="yellow")

table.add_row("main.py", "120", "✅ 通过")
table.add_row("utils.py", "45", "✅ 通过")
table.add_row("tests.py", "89", "❌ 失败")

console.print(table)

# 2. 进度条(替代无聊的百分比日志)
for i in track(range(100), description="处理中..."):
    time.sleep(0.02)  # 模拟耗时操作

# 3. 彩色打印
rprint("[bold green]成功![/] 文件已保存")
rprint("[red]错误:[/] 磁盘空间不足")
rprint("[blue underline]https://example.com[/]")

# 4. 语法高亮
from rich.syntax import Syntax
code = '''
def hello():
    """Greet the world."""
    print("Hello, World!")
'''
syntax = Syntax(code, "python", theme="monokai", line_numbers=True)
console.print(syntax)
💡自动化工具小结

结合本章所学,你可以构建完整的自动化工作流:用 subprocess 调用外部程序, 用 argparse/click 构建命令行界面,用 Pillow 处理图像, 用 rich 美化输出——最后用 APScheduler 定时运行这一切。