Python 常用标准库盘点

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）

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)

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)]

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 -> 北京时间

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"

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

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)

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()