自动化项目最忌“上来就写循环”。正确的流程是“手动跑通 → 拆解步骤 → 逐步脚本化 → 异常兜底”。请严格按以下 6 步执行:
不写一行代码,先手动完成整个任务,记录每一步的输入、输出和耗时。
💡 你无法自动化一个你自己都没跑通的流程。将大任务拆成“打开文件→读取数据→清洗→写入→关闭”等不可再分的原子步骤。
💡 每个原子操作对应一个独立函数,便于测试和复用。收集正常数据、空值、格式错误、超大文件等边界样本。
💡 自动化脚本80%的bug来自没考虑到的脏数据。每完成一个原子操作就运行验证,确认输出符合预期后再进入下一步。
💡 不要一口气写完500行再调试,那是灾难的开始。串联各步骤,添加 try-except、日志记录、重试机制和进度条。
💡 自动化脚本必须能在无人值守时优雅地失败并报告原因。配置定时执行,接入消息通知,确保任务成功/失败都能被感知。
💡 没有监控的自动化 = 薛定谔的自动化。原始数据 → 解析逻辑 → 转换规则 → 输出格式 → 环境/权限,永远从数据源头查起。
目标:掌握HTTP请求、HTML解析、动态页面抓取及数据存储。
import requests
from bs4 import BeautifulSoup
import csv
def fetch_articles(url: str) -> list[dict]:
"""采集文章列表"""
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
articles = []
for item in soup.select("article.post"):
articles.append({
"title": item.select_one("h2.title").get_text(strip=True),
"link": item.select_one("a")["href"],
"date": item.select_one("time")["datetime"]
})
return articles
# 存储到CSV
data = fetch_articles("https://example.com/blog")
with open("articles.csv", "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["title", "link", "date"])
writer.writeheader()
writer.writerows(data)
from playwright.sync_api import sync_playwright
def fetch_dynamic_page(url: str) -> str:
"""处理JS渲染的动态页面"""
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(url, wait_until="networkidle")
# 等待特定元素加载完成
page.wait_for_selector(".content-loaded")
content = page.content()
browser.close()
return content
requests+bs4;需要登录/JS渲染用 Playwright;大规模采集用 Scrapy;API接口优先直接调接口而非解析HTML。
目标:读写Excel、数据清洗、报表自动生成。
import pandas as pd
# 读取 & 清洗
df = pd.read_excel("sales_2024.xlsx", sheet_name="Q1")
df = df.dropna(subset=["订单号"]) # 删除空订单
df["金额"] = df["金额"].astype(float) # 类型转换
df["月份"] = pd.to_datetime(df["日期"]).dt.month # 提取月份
# 聚合分析
summary = df.groupby("月份").agg(
总销售额=("金额", "sum"),
订单数=("订单号", "nunique"),
平均客单价=("金额", "mean")
).round(2)
# 导出带格式的Excel
with pd.ExcelWriter("monthly_report.xlsx", engine="openpyxl") as writer:
summary.to_excel(writer, sheet_name="月度汇总")
df.to_excel(writer, sheet_name="明细数据", index=False)
from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
wb = load_workbook("monthly_report.xlsx")
ws = wb["月度汇总"]
# 设置表头样式
header_fill = PatternFill(start_color="1E3A8A", fill_type="solid")
header_font = Font(color="FFFFFF", bold=True, size=12)
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal="center")
# 自动调整列宽
for col in ws.columns:
max_len = max(len(str(cell.value or "")) for cell in col)
ws.column_dimensions[col[0].column_letter].width = min(max_len + 4, 30)
wb.save("monthly_report_styled.xlsx")
| 场景 | 推荐库 | 说明 |
|---|---|---|
| 数据分析/批量处理 | pandas | 向量化操作,速度快,API简洁 |
| 样式/图表/模板填充 | openpyxl | 精细控制单元格格式、合并、公式 |
| 纯读取大文件 | openpyxl(read_only) | 流式读取,内存占用低 |
| 生成复杂报表 | XlsxWriter | 写入性能极佳,支持图表 |
目标:合同/报告/通知书的批量生成与内容提取。
from docx import Document
from docx.shared import Pt, Inches
from docx.enum.text import WD_ALIGN_PARAGRAPH
doc = Document()
# 标题
title = doc.add_heading("2024年度工作总结", level=0)
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
# 正文段落
doc.add_paragraph("本报告自动生成于2024年12月31日。")
# 表格
table = doc.add_table(rows=4, cols=3, style="Light Grid Accent 1")
headers = ["部门", "营收(万)", "同比增长"]
for i, h in enumerate(headers):
table.rows[0].cells[i].text = h
data = [("技术部", "1200", "+15%"), ("市场部", "800", "+8%"), ("运营部", "600", "+12%")]
for row_idx, row_data in enumerate(data, start=1):
for col_idx, val in enumerate(row_data):
table.rows[row_idx].cells[col_idx].text = val
# 插入图片
doc.add_picture("chart.png", width=Inches(5.5))
doc.save("annual_report.docx")
from docxtpl import DocxTemplate
tpl = DocxTemplate("contract_template.docx")
context = {
"company_name": "XX科技有限公司",
"sign_date": "2024-12-31",
"amount": "¥1,200,000.00",
"items": [
{"name": "软件开发服务", "price": "800,000"},
{"name": "运维支持服务", "price": "400,000"},
]
}
tpl.render(context)
tpl.save("contracts/XX科技_合同.docx")
目标:操控GUI应用、定时执行、跨系统串联。
import pyautogui
import time
# ⚠️ 安全设置:鼠标移到屏幕角落时立即中断
pyautogui.FAILSAFE = True
pyautogui.PAUSE = 0.5 # 每步操作间隔
# 打开ERP系统并导出数据
pyautogui.hotkey('win', 'r')
time.sleep(0.5)
pyautogui.typewrite('erp_client.exe', interval=0.05)
pyautogui.press('enter')
time.sleep(5)
# 点击菜单(使用图像识别比坐标更稳定)
export_btn = pyautogui.locateOnScreen('export_btn.png', confidence=0.9)
if export_btn:
pyautogui.click(pyautogui.center(export_btn))
else:
raise RuntimeError("未找到导出按钮,界面可能已变更")
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
scheduler = BlockingScheduler()
@scheduler.scheduled_job('cron', hour=8, minute=30, day_of_week='mon-fri')
def daily_report():
logger.info(f"开始生成日报 {datetime.now()}")
try:
# 调用你的自动化函数
generate_daily_report()
logger.info("日报生成成功 ✅")
except Exception as e:
logger.error(f"日报生成失败 ❌: {e}")
send_alert(f"日报任务异常: {e}")
scheduler.start()
| 自动化层级 | 工具 | 适用场景 |
|---|---|---|
| API级(最优) | requests / httpx | 有接口的系统间数据同步 |
| 文件级 | pandas / openpyxl / docxtpl | Excel/Word/PDF批量处理 |
| 浏览器级 | Playwright / Selenium | Web系统无API时的操作 |
| 桌面级(最差) | pyautogui / uiautomation | 老旧C/S架构、无其他途径时 |
# ✅ 合规检查清单
ROBOTS_TXT = True # 遵守 robots.txt
RATE_LIMIT = 2 # 每秒最多2次请求
RESPECT_CACHE = True # 缓存已爬页面,避免重复请求
USER_AGENT_IDENTIFY = True # UA中标识身份和联系方式
DATA_MINIMIZATION = True # 只采集必要字段,及时脱敏
LEGAL_REVIEW = True # 商用前务必法务审核
.env 或 config.yaml,绝不硬编码。requirements.txt 或 poetry.lock 锁定精确版本。| 领域 | 首选工具 | 备选/进阶 | 一句话选型理由 |
|---|---|---|---|
| 静态爬虫 | requests + bs4 | httpx + lxml | 生态最全,文档最多 |
| 动态爬虫 | Playwright | Selenium / DrissionPage | 比Selenium快3倍,API更现代 |
| 大规模采集 | Scrapy | Crawlee | 异步框架,内置去重/限速/管道 |
| Excel处理 | pandas | openpyxl / XlsxWriter | 数据处理王者,一行顶百行 |
| Word生成 | python-docx | docxtpl | 模板渲染用docxtpl,从零构建用docx |
| PDF处理 | pdfplumber | PyMuPDF / reportlab | 提取文本表格最准 |
| 桌面RPA | pyautogui | uiautomation / pywinauto | 跨平台简单操作用autogui,Windows深度操作用uiautomation |
| 任务调度 | APScheduler | cron / Celery Beat | 轻量级Python原生调度 |