9个Python自动化神器:让我感觉自己像个黑客!
这种活要是手点,我第一眼就不太信。不是做不了,是人会犯困,手会点错,最后锅还得程序员背。
我后来习惯先写个 Python 小脚本。几十行代码跑起来,屏幕刷刷输出,文件自动归档,接口自动探活,Excel 自动生成结果。那一刻确实有点像电影里的黑客,当然,干的是正经活。
第一个我离不开的是 pathlib。
以前写文件路径,我最烦一堆字符串拼接,Windows 一个斜杠,Linux 一个斜杠,脚本换台机器就抽风。pathlib 用起来顺手很多。
比如清理下载目录,把不同后缀的文件丢到不同文件夹:
from pathlib import Path
from shutil import move
box = Path.home() / "Downloads"
rules = {
"images": {".png", ".jpg", ".jpeg"},
"docs": {".pdf", ".docx", ".xlsx"},
"logs": {".log", ".txt"},
"zip": {".zip", ".rar", ".7z"},
}
for f in box.iterdir():
ifnot f.is_file():
continue
ext = f.suffix.lower()
target_name = next((k for k, v in rules.items() if ext in v), "others")
target_dir = box / target_name
target_dir.mkdir(exist_ok=True)
new_path = target_dir / f.name
print(f"move {f.name} -> {target_name}")
move(str(f), str(new_path))
这种脚本没什么技术含量,但很救命。尤其是你刚接手一台乱七八糟的测试机,先跑一遍,目录至少能看了。
第二个是 os 和 subprocess。
os 管环境、目录、进程信息,subprocess 负责调系统命令。很多自动化不是 Python 自己干完,而是 Python 把一堆命令串起来。
我一般用它做服务巡检:
import subprocess
from datetime import datetime
services = ["nginx", "mysql", "redis-server"]
for name in services:
cmd = ["systemctl", "is-active", name]
ret = subprocess.run(cmd, capture_output=True, text=True)
status = ret.stdout.strip()
line = f"{datetime.now():%F %T}{name:<12}{status}"
print(line)
if status != "active":
print(f"[WARN] {name} 状态不对,先别急着重启,去看日志")
这里我不会上来就自动重启。线上脚本最怕“太勤快”,它把故障现场给你抹了。先报警,先留痕,再决定怎么处理。
第三个是 requests。
接口探活、批量调用、下载文件、调内部平台,它基本都能干。拿它查接口健康状态,比手动点浏览器靠谱。
import requests
from time import perf_counter
apis = [
"https://example.com/api/health",
"https://example.com/api/order/status",
]
for url in apis:
start = perf_counter()
try:
r = requests.get(url, timeout=3)
cost = (perf_counter() - start) * 1000
print(f"{r.status_code}{cost:.1f}ms {url}")
except requests.RequestException as e:
print(f"FAIL {url}{type(e).__name__}")
我排接口慢的时候,一般不会先翻业务代码。先看状态码、耗时、超时位置,再去查 trace。方向错了,代码翻一天也没用。
第四个是 openpyxl。
这玩意儿对付 Excel 很实在。运营、财务、产品发来的表,格式经常不讲武德。用 Python 读一遍,比你手筛稳得多。
比如把金额异常的数据挑出来:
from openpyxl import load_workbook, Workbook
src = load_workbook("orders.xlsx")
sheet = src.active
out = Workbook()
bad = out.active
bad.append(["order_id", "user_id", "amount", "reason"])
for row in sheet.iter_rows(min_row=2, values_only=True):
order_id, user_id, amount = row[0], row[1], row[4]
if amount isNone:
bad.append([order_id, user_id, amount, "金额为空"])
elif amount < 0or amount > 50000:
bad.append([order_id, user_id, amount, "金额范围异常"])
out.save("bad_orders.xlsx")
这种脚本我建议每次都输出一个新文件,别直接改原表。原表是证据,改坏了不好复盘。
第五个是 pandas。
openpyxl 适合精细操作 Excel,pandas 适合批量清洗数据。几万行 CSV,字段补齐、去重、分组统计,用它很快。
import pandas as pd
df = pd.read_csv("access.log.csv")
df["path"] = df["url"].str.split("?").str[0]
slow = (
df[df["cost_ms"] > 1000]
.groupby("path")
.agg(count=("path", "size"), avg_cost=("cost_ms", "mean"))
.sort_values("count", ascending=False)
)
slow.to_csv("slow_api_top.csv")
慢接口别只盯平均耗时,我更喜欢先看“慢的次数”。偶发慢和持续慢,不是一类问题。
第六个是 watchdog。
它可以监听目录变化。文件一落地,脚本马上处理。做日志采集、图片压缩、导入文件检查,很舒服。
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
import time
classImportWatcher(FileSystemEventHandler):
defon_created(self, event):
p = Path(event.src_path)
if p.suffix.lower() == ".csv":
print(f"新导入文件: {p.name},准备校验")
observer = Observer()
observer.schedule(ImportWatcher(), "./import_box", recursive=False)
observer.start()
try:
whileTrue:
time.sleep(1)
finally:
observer.stop()
observer.join()
这里有个坑,文件刚创建不代表写完了。真上生产,要加文件大小稳定判断,不然你可能读到半截 CSV。
第七个是 schedule。
轻量定时任务不用一上来就搞 Airflow。每天清日志、每小时查接口、每十分钟扫目录,schedule 就够。
import schedule
import time
from pathlib import Path
defclean_tmp():
tmp = Path("./tmp")
for f in tmp.glob("*.cache"):
if f.stat().st_size == 0:
f.unlink()
print("delete empty cache:", f.name)
schedule.every(10).minutes.do(clean_tmp)
whileTrue:
schedule.run_pending()
time.sleep(1)
它适合小脚本,不适合强一致任务。比如扣款、发券这种,别靠它硬扛。
第八个是 argparse。
脚本一旦给别人用,就别让人改源码里的变量。参数写清楚,错误少一半。
import argparse
from pathlib import Path
parser = argparse.ArgumentParser()
parser.add_argument("--dir", required=True, help="要扫描的目录")
parser.add_argument("--suffix", default=".log", help="文件后缀")
args = parser.parse_args()
for f in Path(args.dir).glob(f"*{args.suffix}"):
print(f.name)
能命令行传参的脚本,才像个工具。否则就是一次性草稿。
第九个是 rich。
它不是自动化核心,但能让输出像样。脚本跑半天没反馈,人会慌。加个表格、进度条、颜色提示,排障体验完全不一样。
from rich.console import Console
from rich.table import Table
console = Console()
table = Table(title="接口巡检结果")
table.add_column("服务")
table.add_column("状态")
table.add_column("耗时")
table.add_row("order-api", "OK", "83ms")
table.add_row("pay-api", "WARN", "1320ms")
table.add_row("user-api", "OK", "45ms")
console.print(table)
别小看输出格式。线上排查时,一堆日志糊在一起,眼睛先废。
这些库单独看都不神秘,但组合起来就很猛。
pathlib 扫文件,watchdog 监听目录,pandas 清洗数据,requests 调接口,schedule 定时跑,最后用 rich 打出结果。一个下午能省掉以后很多重复劳动。
我现在写自动化脚本有个习惯:先别追求大而全,先把最烦、最容易点错的那一步干掉。
比如先自动归档文件。
再自动校验字段。
再自动生成异常报告。
最后才考虑定时和告警。
工具写小一点,坏了也好查。脚本一旦写成“平台”,维护成本就上来了。Python 最爽的地方不在于它能写多复杂,而是你被琐事烦到不行的时候,二十分钟能掏出一把顺手的小刀。