前言 Python “自带电池”——标准库覆盖了运维脚本 80% 的需求。这篇盘点我使用频率最高的那些，每个都给实战片段。
一、os / sys / pathlib：系统交互 import os, sys from pathlib import Path # 环境与退出 os.environ.get("HOME", "/root") # 读环境变量（带默认值） os.getpid(), os.getcwd() sys.exit(1) # 带状态码退出（脚本规范） # pathlib 三板斧（上篇讲过, 这里补充） for f in Path("/var/log").rglob("*.log"): # 递归找 print(f, f.stat().st_size) # 执行外部命令的正确姿势见 subprocess（别用 os.system） 二、collections：数据结构增强 from collections import Counter, defaultdict, deque, OrderedDict # Counter: 计数器之王 c = Counter(["go", "py", "go", "sh", "go"]) c.most_common(2) # [('go', 3), ('py', 1)] c["go"] # 3（不存在的 key 返回 0, 不报错） # defaultdict: 消灭 if key not in groups = defaultdict(list) for name, dept in [("a", "ops"), ("b", "dev"), ("c", "ops")]: groups[dept].append(name) # {'ops': ['a', 'c'], 'dev': ['b']} logs = defaultdict(int) # 默认 0: 直接 += 1 不会 KeyError logs["error"] += 1 # deque: 双端队列（队列/栈/滑动窗口） q = deque(maxlen=5) for i in range(10): q.append(i) # 自动挤掉最老的 # deque([5, 6, 7, 8, 9]) q.appendleft(0) # O(1) 头部插入, list 是 O(n) 三、itertools：迭代器魔法 from itertools import chain, islice, groupby, product, combinations # chain: 拼接多个可迭代对象 list(chain([1, 2], [3], range(4, 6))) # [1, 2, 3, 4, 5] # islice: 切迭代器（不能下标的对象） list(islice(range(100), 5, 10)) # [5, 6, 7, 8, 9] # product: 笛卡尔积（替代多层 for） for env, svc in product(["prod", "staging"], ["web", "api"]): print(f"{env}-{svc}") # groupby: 分组（注意: 先排序再分组!） data = sorted([("a", 1), ("b", 2), ("a", 3)], key=lambda x: x[0]) for k, g in groupby(data, key=lambda x: x[0]): print(k, list(g)) # a [('a',1),('a',3)] / b [('b',2)] 四、datetime：时间处理 from datetime import datetime, timedelta, timezone now = datetime.now() utc = datetime.now(timezone.utc) # 时区 aware（推荐!） # 格式化与解析 now.strftime("%Y-%m-%d %H:%M:%S") # '2024-06-15 10:30:00' datetime.strptime("2024-06-15", "%Y-%m-%d") # 字符串 -> datetime datetime.fromisoformat("2024-06-15T10:00") # ISO 格式直解 # 运算 yesterday = now - timedelta(days=1) (now - yesterday).days # 1 now.timestamp() # -> Unix 时间戳 datetime.fromtimestamp(1718419200) # <- Unix 时间戳 # 时区转换 tz_sh = timezone(timedelta(hours=8)) utc.astimezone(tz_sh) # UTC -> 北京时间 五、subprocess：执行命令（重点） import subprocess # ✅ 推荐写法: run + capture（3.5+） r = subprocess.run( ["df", "-h"], # 列表形式（不经 shell, 更安全） capture_output=True, text=True, timeout=10, ) print(r.returncode) # 0 print(r.stdout) # 带 shell 管道时才用 shell=True（注意注入风险!） r = subprocess.run("df -h | grep data", shell=True, capture_output=True, text=True) # 失败抛异常（写脚本报错即停） r = subprocess.run(["ls", "/nope"], check=True, capture_output=True, text=True) # 实战: 检查服务是否存活 def service_active(name: str) -> bool: r = subprocess.run(["systemctl", "is-active", name], capture_output=True, text=True) return r.stdout.strip() == "active" 六、argparse：像样的命令行参数 import argparse p = argparse.ArgumentParser(description="主机巡检工具") p.add_argument("hosts", nargs="+", help="目标主机列表") p.add_argument("-p", "--port", type=int, default=22) p.add_argument("-v", "--verbose", action="store_true") args = p.parse_args() # python check.py web01 web02 -p 2222 -v # args.hosts=['web01','web02'] args.port=2222 args.verbose=True 七、json / re / hashlib / secrets：日常刚需 import json, re, hashlib, secrets # json（上篇详述） json.dumps({"k": "值"}, ensure_ascii=False, indent=2) # re: 三个够用的函数 re.search(r"(\d+\.\d+\.\d+\.\d+)", text) # 找第一处（返回 match） re.findall(r"ERROR.*", log) # 找全部 re.sub(r"\s+", " ", s) # 替换 # 预编译（循环里高频使用时） IP = re.compile(r"^(?:\d{1,3}\.){3}\d{1,3}$") IP.match("10.0.0.1") # hashlib: 校验文件 h = hashlib.sha256() for chunk in iter(lambda: open("app.tar.gz","rb").read(8192), b""): h.update(chunk) print(h.hexdigest()) # secrets: 生成安全随机串（密码/Token, 别用 random!） secrets.token_urlsafe(32) 八、logging：别再 print 了 import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) log = logging.getLogger(__name__) log.info("开始巡检 %s", "web01") # 惰性格式化, 占位符写法 try: 1 / 0 except ZeroDivisionError: log.exception("计算失败") # 自动带堆栈 九、综合实战：一行式巡检脚本 #!/usr/bin/env python3 """快速体检: 磁盘/内存/负载""" import subprocess, shutil, os def sh(cmd: str) -> str: r = subprocess.run(cmd, shell=True, capture_output=True, text=True) return r.stdout.strip() def check_disk(threshold=80): for line in sh("df -h --output=pcent,target").splitlines()[1:]: pct, mount = line.split() if int(pct.rstrip("%")) > threshold: print(f"[告警] {mount} 使用率 {pct}") def check_load(): cores = os.cpu_count() load = float(sh("cat /proc/loadavg").split()[0]) status = "告警" if load > cores else "正常" print(f"[{status}] load {load} / {cores} cores") def check_mem(): total, _, free, _, _, avail = sh("free -b").splitlines()[1].split()[1:] print(f"[信息] 可用内存 {int(avail)/2**30:.1f}G") if __name__ == "__main__": check_disk(); check_load(); check_mem() 零第三方依赖，拷到任何有 Python3 的机器就能跑。
...