凌晨两点,运营又甩过来一个包:3000 个用户目录,按日期归档,顺手把异常日志挑出来,再补一份 Excel。 这种活儿你要是手点,点到天亮都未必完。 但 Python 一上,味道就变了。不是会不会写业务的问题,是很多脏活一旦自动化,整个人都会产生一种“我是不是偷偷开挂了”的错觉。
先说清楚,这里不整那种教材味的 demo。都是平时真能落地的东西。
我现在看到有人还在写 "C:\\a\\b\\" + name,第一反应就是这代码后面八成还有坑。
from pathlib import Pathbase = Path("/data/order_logs")today_dir = base / "2026-03-18"for f in today_dir.glob("*.log"):if f.stat().st_size == 0: print("空日志:", f.name)
这个东西最大的好处不是“优雅”,是少出错。路径拼接、遍历、过滤,顺手就做了。
归档、备份、迁移,这种操作最怕漏文件、覆盖错目录。
import shutilfrom pathlib import Pathsrc = Path("/data/report")bak = Path("/data/report_bak")bak.mkdir(exist_ok=True)for f in src.glob("*.xlsx"): target = bak / f"{f.stem}_bak{f.suffix}" shutil.copy2(f, target)
copy2 会尽量保留文件时间这些元信息。你别小看这个,很多排查现场最后就靠时间戳对链路。
批量调接口、补数据、重试失败任务,十次里有八次离不开它。 我一般不会上来就并发轰,先把 session、超时、重试想清楚。
import requestssession = requests.Session()session.headers.update({"X-Token": "test-token"})users = [1001, 1002, 1003]for uid in users:try: resp = session.get(f"https://api.example.com/user/{uid}", timeout=3 ) print(uid, resp.status_code, resp.text[:60])except requests.RequestException as e: print("请求失败", uid, e)
超时一定要带。线上那种“脚本跑着跑着没声了”,很多就是卡死在这里。
不是所有活都值得上 Airflow。 有些脚本就是每天 9 点跑一下,把昨天失败单捞出来重试,schedule 已经够了。
import scheduleimport timedefretry_failed_orders(): print("开始补偿失败订单...")schedule.every().day.at("09:00").do(retry_failed_orders)whileTrue: schedule.run_pending() time.sleep(1)
小团队里,这种东西很实用。别一上来就把架子搭得像航天工程。
有些活,shell 本来就更顺手。 比如 grep 日志、压缩文件、调用已有脚本,Python 负责编排,系统命令负责干脏活。
import subprocesscmd = "grep 'ERROR' /data/app.log | tail -n 20"result = subprocess.run(cmd, shell=True, capture_output=True, text=True)print(result.stdout)
这玩意儿特别适合排障脚本。 我一般会拿它先把现场证据抠出来,再决定要不要继续分析。
日志里最烦的不是多,是脏。 你要从一坨文本里抓订单号、IP、traceId,不上正则纯属给自己找事。
import reline = '2026-03-18 10:11:22 ERROR traceId=9fa21 orderId=A20260318001 timeout'm = re.search(r"traceId=(\w+).*orderId=([A-Z0-9]+)", line)if m: print("traceId:", m.group(1)) print("orderId:", m.group(2))
正则不用搞太花,够准就行。 现场脚本最怕炫技,能抓到值比写得漂亮重要。
csv / openpyxl:表格活,别再复制粘贴了
很多“数据处理”其实没什么技术含量,就是重复。 把日志筛出来,吐成 Excel 给运营、测试、财务,他们立刻就能用。
from openpyxl import Workbookrows = [ ["order_id", "status", "remark"], ["A1001", "fail", "支付超时"], ["A1002", "success", ""],]wb = Workbook()ws = wb.activews.title = "result"for row in rows: ws.append(row)wb.save("order_result.xlsx")
这种脚本写一次,后面你就不想再手工整理表格了。
concurrent.futures:批量 I/O 提速很明显
批量请求接口、扫描文件、检查端口,这种 I/O 型任务串行跑太亏。
from concurrent.futures import ThreadPoolExecutorimport requestsdefcheck(url):try: r = requests.get(url, timeout=2)return url, r.status_codeexcept Exception:return url, "DOWN"urls = ["https://a.example.com/health","https://b.example.com/health","https://c.example.com/health",]with ThreadPoolExecutor(max_workers=5) as pool:for item in pool.map(check, urls): print(item)
线程数别瞎拉满。你是做检查,不是做压测。这个边界感,很多脚本作者没有。
这个很像“埋伏”。 目录里只要新文件一落地,脚本立刻处理,特别适合导入、转码、上传这种场景。
from watchdog.observers import Observerfrom watchdog.events import FileSystemEventHandlerimport timeclassNewFileHandler(FileSystemEventHandler):defon_created(self, event):ifnot event.is_directory: print("发现新文件:", event.src_path)observer = Observer()observer.schedule(NewFileHandler(), "/data/incoming", recursive=False)observer.start()try:whileTrue: time.sleep(1)finally: observer.stop() observer.join()
很多人把自动化理解成“定时跑”。 其实更狠的是“事件一来,马上动”。
说到底,Python 自动化最爽的地方,不是代码短,也不是库多。 而是你会越来越懒得重复劳动。