进阶篇 / 第 14 章

第 14 章 · 常用标准库巡礼

标准库地图与学习方法

标准库模块有数百个，但真正高频的无非下面几类。先建立全局地图，再逐个精讲：

collections：内置容器的加强版

collections 提供了一组比内置 list/dict 更贴合具体场景的容器。记不住全部没关系，先掌握下面这几个高频的。

2.1 defaultdict / Counter / OrderedDict

from collections import defaultdict, Counter, deque, OrderedDict

# defaultdict：不存在时用工厂函数造默认值
scores = defaultdict(list)
for name, score in [("a", 90), ("b", 80), ("a", 95)]:
    scores[name].append(score)          # 不再需要 setdefault
print(dict(scores))
print("默认值类型:", type(defaultdict(int)["missing"]))

# Counter：计数 / 最常见 / 加法
cnt = Counter("mississippi")
print(cnt.most_common(3), cnt["s"], cnt.most_common(1)[0][1])
a = Counter("ab") + Counter("bc")      # 相加 = 逐键累加
print(a, Counter("ab") - Counter("b")) # 减法 = 只保留正数

# OrderedDict：记住插入顺序（普通 dict 3.7+ 已有序，但 popitem 语义不同）
od = OrderedDict([("z", 1), ("a", 2)])
od.move_to_end("z")
print(list(od.keys()), od.popitem(last=False))

输出：

{'a': [90, 95], 'b': [80]}
默认值类型: <class 'int'>
[('s', 4), ('i', 4), ('p', 2)] 4 4
Counter({'b': 2, 'a': 1, 'c': 1}) Counter({'a': 1})
['a', 'z'] ('a', 2)

2.2 deque：双端队列

做队列（BFS）、滑动窗口、缓存淘汰时，list.pop(0) 是 O(n) 的灾难。deque 两端都是 O(1)：

deque_demo.py

from collections import deque

q = deque([1, 2, 3])
q.append(4)          # 右端
q.appendleft(0)      # 左端
print(q, len(q))
print(q.popleft(), q.pop())

# 滑动窗口最大值（BFS 场景的标准结构）
data = [1, 3, -1, -3, 5, 3, 6, 7]
window = 3
dq: deque[int] = deque()
result = []
for i, v in enumerate(data):
    while dq and data[dq[-1]] <= v:
        dq.pop()
    dq.append(i)
    if dq[0] <= i - window:
        dq.popleft()
    if i >= window - 1:
        result.append(data[dq[0]])
print("滑动窗口最大值:", result)     # [3, 3, 5, 5, 6, 7]

# 有界 deque：天然实现 LRU 上限
cache: deque[str] = deque(maxlen=3)
for ch in "abcdef":
    cache.append(ch)
print(list(cache))

输出：

deque([0, 1, 2, 3, 4]) 5
0 4
滑动窗口最大值: [3, 3, 5, 5, 6, 7]
['d', 'e', 'f']

2.3 namedtuple / ChainMap / UserDict

from collections import namedtuple, ChainMap, UserDict

# namedtuple：用名字访问的元组
Point = namedtuple("Point", ["x", "y", "label"])
p = Point(1, 2, "起点")
print(p, p.x, p._asdict(), p._replace(x=9))

# ChainMap：多个 dict 合并视图，后者覆盖前者
defaults = {"theme": "light", "lang": "zh"}
user = {"theme": "dark"}
cfg = ChainMap(user, defaults)
print(cfg["theme"], cfg["lang"], cfg.maps)
print(dict(cfg))                # 拍平成普通 dict

# UserDict：给所有 dict-like 行为套一层壳（便于加校验/默认值）
class CaseInsensitiveDict(UserDict):
    def __init__(self, *args, **kw):
        super().__init__(*args, **kw)
        self._store = {k.lower(): (k, v) for k, v in self.data.items()}
    def __setitem__(self, k, v):
        self._store[k.lower()] = (k, v)
    def __getitem__(self, k):
        return self._store[k.lower()][1]

d = CaseInsensitiveDict({"Name": "张三"})
print(d["name"], dict(d.data))

输出：

Point(x=1, y=2, label='起点') 1 {'x': 1, 'y': 2, 'label': '起点'} Point(x=9, y=2, label='起点')
dark zh [{'theme': 'dark'}, {'theme': 'light', 'lang': 'zh'}]
{'theme': 'dark', 'lang': 'zh'}
张三 {'Name': '张三'}

2.4 容器选型速查表

itertools：惰性迭代流水线

itertools 提供三类工具：无限迭代器（count/cycle/repeat）、有限终止迭代器（takewhile/dropwhile/islice/groupby）、组合器（product/permutations/combinations/chain）。全部是惰性（lazy）的，不占内存。

3.1 无限迭代器与 islice

from itertools import count, cycle, repeat, islice, takewhile, dropwhile

# count(start, step) —— 无限计数
print(list(islice(count(0, 3), 5)))            # [0, 3, 6, 9, 12]

# cycle —— 无限循环一个可迭代对象
letters = cycle("abc")
print("".join(islice(letters, 7)))             # abcabca

# repeat —— 重复单个值
print(list(islice(repeat("x"), 3)), list(repeat(0, 3)))

# takewhile：条件成立前一直取；dropwhile：丢掉开头一段
nums = [2, 4, 6, 8, 3, 1, 5]
print(list(takewhile(lambda x: x % 2 == 0, nums)))   # [2,4,6,8]
print(list(dropwhile(lambda x: x % 2 == 0, nums)))   # [3,1,5]

# islice：切一段（不真正切片列表）
data = range(100)
print(list(islice(data, 10, 20, 2)))           # [10, 12, 14, 16, 18]

输出：

[0, 3, 6, 9, 12]
abcabca
['x', 'x', 'x'] [0, 0, 0]
[2, 4, 6, 8]
[3, 1, 5]
[10, 12, 14, 16, 18]

3.2 groupby：分组处理（最容易用错的一个）

from itertools import groupby

# 关键前提：数据必须已按分组键排序！
records = [
    ("fruit", "apple"), ("fruit", "banana"),
    ("veg", "carrot"),
    ("fruit", "cherry"),          # 未排序 -> 会形成新组
]
for key, group in groupby(sorted(records, key=lambda r: r[0]), key=lambda r: r[0]):
    print(key, [v for _, v in group])

from collections import defaultdict
d = defaultdict(list)
for r in records:
    d[r[0]].append(r[1])
print(dict(d))

输出：

fruit ['apple', 'banana', 'cherry']
veg ['carrot']
{'fruit': ['apple', 'banana', 'cherry'], 'veg': ['carrot']}

3.3 组合器

from itertools import product, permutations, combinations, combinations_with_replacement, chain, zip_longest

print(list(product([1, 2], "ab")))                    # 笛卡尔积
print(list(permutations([1, 2, 3], 2)))                # 排列（不重复元素）
print(list(combinations([1, 2, 3], 2)))                # 组合（不计顺序）
print(list(combinations_with_replacement([1, 2], 2)))   # 可重复组合
print(list(product([0, 1], repeat=3)))                 # 二进制计数

a, b = [1, 2], ["x", "y", "z"]
print(list(chain(a, b)))                               # 串接多个可迭代
print(list(zip_longest(a, b, fillvalue="?")))          # 按最长补齐

输出：

[(1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')]
[(1, 2), (1, 3), (2, 1), (2, 3), (3, 1), (3, 2)]
[(1, 2), (1, 3), (2, 3)]
[(1, 1), (1, 2), (2, 1), (2, 2)]
[(0, 0, 0), (0, 0, 1), (0, 1, 0), ...]
[1, 2, 'x', 'y', 'z']
[(1, 'x'), (2, 'y'), (None, 'z')]

3.4 惰性的真正价值：处理无限/超大数据流

from itertools import islice, count

def primes():
    """惰性生成素数：占用恒定内存。"""
    seen = set()
    for n in count(2):
        if all(n % p for p in seen if p * p <= n):
            seen.add(n)
            yield n

# 只取前 10 个素数，内存恒定
first_ten = list(islice(primes(), 10))
print("前 10 素数:", first_ten)

# 惰性求值：生成器不执行，直到被消费
lazy = (x * x for x in range(1_000_000_000))
print("未消费时内存恒定，取前 5 个:", list(islice(lazy, 5)))

输出：

前 10 素数: [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]
未消费时内存恒定，取前 5 个: [1, 4, 9, 16, 25]

3.5 functools：缓存、偏函数与归约

import functools
import time

@functools.lru_cache(maxsize=128)
def slow_square(n: int) -> int:
    """带缓存的慢函数（第二次起瞬时返回）。"""
    time.sleep(0.01)
    return n * n

t0 = time.perf_counter()
print(slow_square(12), slow_square(12), slow_square.cache_info())
print(f"两次调用耗时 {time.perf_counter() - t0:.3f}s")   # 只有一次真的 sleep

# functools.cache 是 lru_cache(maxsize=None) 的语法糖
@functools.cache
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)
print(fib(50), fib.cache_info().currsize)

# partial：偏函数，预先绑定部分参数
def greet(greeting: str, name: str) -> str:
    return f"{greeting}, {name}!"

hello = functools.partial(greet, "Hello")
print(hello("Bob"), hello("Alice"))

# reduce + 初始值
print(functools.reduce(lambda x, y: x * y, [1, 2, 3, 4], 1))
# 单参数用 map 更直白：把每项平方
print(list(map(lambda x: x**2, [1, 2, 3])))

# wraps 保留元信息（写装饰器必备）
def timed(fn):
    @functools.wraps(fn)
    def inner(*a, **k):
        t = time.perf_counter()
        r = fn(*a, **k)
        print(f"  {fn.__name__} 耗时 {time.perf_counter() - t:.4f}s")
        return r
    return inner

@timed
def work():
    time.sleep(0.01)
    return "done"
print(work(), work.__name__)

输出：

144 144 CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)
两次调用耗时 0.011s
12586269025 CacheInfo(hits=48, misses=50, maxsize=None, currsize=50)
Hello, Bob! Hello, Alice!
24
[1, 4, 9]
  work 耗时 0.0102s
done work

datetime：时间处理

datetime 模块的核心类型是 datetime（日期+时间点）、date（只有日期）、time、timedelta（时间差）、tzinfo（时区）。

4.1 创建、取字段、格式化

from datetime import datetime, date, timedelta, timezone

now = datetime.now()
print("现在:", now.strftime("%Y-%m-%d %H:%M:%S"))
print("年月日时分秒:", now.year, now.month, now.day, now.hour, now.minute, now.second)
print("星期(0=周一):", now.weekday(), "| ISO 周:", now.isocalendar())
print("日期:", date.today())

# 解析字符串 -> datetime
d = datetime.strptime("2024-03-15 08:30:00", "%Y-%m-%d %H:%M:%S")
print("解析:", d, "| ISO:", d.isoformat())

# 时间差：加减
print("+90分钟:", d + timedelta(minutes=90))
print("+1天2小时:", d + timedelta(days=1, hours=2))
print("间隔:", datetime(2024, 3, 15) - datetime(2024, 3, 1))

# 时区：aware vs naive
utc_now = datetime.now(timezone.utc)
bj = utc_now.astimezone(timezone(timedelta(hours=8)))
print("UTC:", utc_now.strftime("%H:%M"), "| 北京:", bj.strftime("%H:%M"))

输出（示例）：

现在: 2026-10-18 11:04:22
年月日时分秒: 2026 10 18 11 4 22
星期(0=周一): 5 | ISO 周: datetime.IsoCalendarDate(year=2026, week=42, weekday=6)
日期: 2026-10-18
解析: 2024-03-15 08:30:00 | ISO: 2024-03-15T08:30:00
+90分钟: 2024-03-15 10:00:00
+1天2小时: 2024-03-16 10:30:00
间隔: 14 days, 0:00:00
UTC: 03:04 | 北京: 11:04

4.2 类型化日期：掌握 datetime.date

from datetime import date, timedelta

birth = date(1995, 6, 15)
today = date.today()
age_days = (today - birth).days
age_years = today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
print(f"活了 {age_days} 天，约 {age_years} 岁")
print("闰年判断:", date(2024, 2, 29).timetuple().tm_yday == 60)
print("月末:", birth.replace(day=1), date(2024, 1, 31) + timedelta(days=1))

# 工作日计算（跳过周末）
def add_business_days(start: date, n: int) -> date:
    cur, step = start, 1 if n >= 0 else -1
    remaining = abs(n)
    while remaining:
        cur += timedelta(days=step)
        if cur.weekday() < 5:
            remaining -= 1
    return cur

print("5 个工作日后:", add_business_days(date(2024, 3, 11), 5))

输出（示例）：

活了 11408 天，约 31 岁
闰年判断: True
月末: 1995-06-01 2024-03-01
5 个工作日后: 2024-03-18

4.3 时间戳与 perf_counter

from datetime import datetime
import time

# 时间戳：秒级 Unix 时间（浮点含小数），适合与网络接口交互
ts = datetime.now().timestamp()
print("时间戳:", round(ts, 3))
print("还原:", datetime.fromtimestamp(ts))

# 计时用 perf_counter（单调递增、精度高），不要用 time.time()
t0 = time.perf_counter()
for _ in range(100_000):
    pass
print(f"perf_counter 耗时 {time.perf_counter() - t0:.4f}s")

json 与 csv：数据序列化

5.1 json

import json
from datetime import date

data = {"name": "张三", "age": 18, "tags": ["a", "b"], "nested": {"ok": True}}

# 序列化：ensure_ascii=False 保留中文；indent 美化
s = json.dumps(data, ensure_ascii=False, indent=2)
print(s)

# 反序列化
back = json.loads(s)
print(back["nested"]["ok"], type(back["name"]))

# 紧凑（网络传输）
compact = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
print(compact)

# 文件读写
with open("t.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)
with open("t.json", "r", encoding="utf-8") as f:
    print("从文件读回:", json.load(f)["name"])

# JSON 没有 date 类型 -> 用 default=str 兜底转字符串
print(json.dumps({"d": date(2024, 3, 15)}, default=str))

输出：

{
  "name": "张三",
  "age": 18,
  "tags": [
    "a",
    "b"
  ],
  "nested": {
    "ok": true
  }
}
True <class 'str'>
{"name":"张三","age":18,"tags":["a","b"],"nested":{"ok":true}}
从文件读回: 张三
{"d": "2024-03-15"}

5.2 CSV

csv_demo.py

import csv
import io

rows = [
    {"id": "1", "name": "张三", "city": "北京"},
    {"id": "2", "name": "李四", "city": "上海"},
]

# 写 CSV（字典的行）
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=["id", "name", "city"])
writer.writeheader()
writer.writerows(rows)
csv_text = buf.getvalue()
print(csv_text)

# 读 CSV
reader = csv.DictReader(io.StringIO(csv_text))
print([dict(r) for r in reader])

# 写列表的行（无表头）
buf2 = io.StringIO()
csv.writer(buf2).writerows([[1, "x", True], [2, "y", False]])
print(repr(buf2.getvalue()))

输出：

id,name,city
1,张三,北京
2,李四,上海

[{'id': '1', 'name': '张三', 'city': '北京'}, {'id': '2', 'name': '李四', 'city': '上海'}]
'1,x,True\r\n2,y,False\r\n'

os 与 pathlib：文件系统

现代 Python 推荐用 pathlib（面向对象）而非 os.path（字符串拼接）。但 os 的进程/环境相关功能仍必须用它。

6.1 pathlib：现代路径处理

from pathlib import Path

base = Path("project")
(base / "src" / "main.py").parent.mkdir(parents=True, exist_ok=True)
(base / "src" / "main.py").write_text("print('hi')\n", encoding="utf-8")
(base / "data.txt").write_text("内容", encoding="utf-8")

p = base / "src" / "main.py"
print("名字/后缀/无后缀:", p.name, p.suffix, p.stem)
print("父目录:", p.parent, "| 各部分:", p.parts)
print("是否绝对:", p.is_absolute())
print("内容:", p.read_text(encoding="utf-8").strip())

# 遍历与 glob
for f in sorted(base.rglob("*")):
    print("  ", f, "dir" if f.is_dir() else f"file {f.stat().st_size}B")

print("匹配:", sorted(str(x) for x in base.rglob("*.py")))
print("目录树:", sorted(str(x.relative_to(base)) for x in base.rglob("*.py")))

# 重命名 / 复制 / 删除
copy = p.with_name("main_copy.py")
p.rename(copy)
copy.unlink()
print("删除后还存在:", copy.exists())
import shutil
shutil.rmtree(base)                       # 递归删目录
print("已清理:", not base.exists())

输出：

名字/后缀/无后缀: main.py .py main
父目录: project/src | 各部分: ('project', 'src', 'main.py')
是否绝对: False
内容: print('hi')
   project dir
   project/data.txt file 6B
   project/src dir
   project/src/main.py file 12B
匹配: ['project/src/main.py']
目录树: ['src/main.py']
删除后还存在: False
已清理: True

6.2 os：环境、路径与权限

import os
import sys
import tempfile

# 环境变量
os.environ["MY_APP"] = "demo"
print("读环境变量:", os.environ.get("MY_APP"), "| 默认值:", os.getenv("NOPE", "fallback"))

# 目录操作
print("cwd:", os.getcwd())
print("分隔符:", os.sep, "| 路径拼接:", os.path.join("a", "b", "c.txt"))
print("是否存在:", os.path.exists("."), os.path.isdir("."), os.path.isfile("nope"))

# 遍历目录
entries = list(os.scandir("."))[:3]
print("scandir 示例:", [e.name + ("/" if e.is_dir() else "") for e in entries])

# 临时文件/目录（测试与脚本必备）
with tempfile.TemporaryDirectory() as tmp:
    with open(os.path.join(tmp, "x.txt"), "w") as f:
        f.write("temp")
    print("临时目录内容:", os.listdir(tmp))
print("临时目录已自动清理")

# 进程信息
print("pid:", os.getpid(), "| argv:", sys.argv[0].endswith(".py"))
print("cpu_count:", os.cpu_count())

输出：

读环境变量: demo | 默认值: fallback
分隔符: / | 路径拼接: a/b/c.txt
是否存在: True True False
临时目录内容: ['x.txt']
pid: 12345 | argv: True
cpu_count: 8

6.3 shutil：文件与目录的批量操作

import shutil
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory() as tmp:
    src = Path(tmp) / "src"
    src.mkdir()
    (src / "a.txt").write_text("A", encoding="utf-8")
    (src / "sub").mkdir()
    (src / "sub" / "b.txt").write_text("B", encoding="utf-8")

    # 复制文件 / 递归复制目录
    shutil.copy(src / "a.txt", Path(tmp) / "a_copy.txt")
    shutil.copytree(src, Path(tmp) / "backup")
    print("备份内容:", sorted(p.name for p in (Path(tmp) / "backup").rglob("*.txt")))

    # 移动
    shutil.move(str(Path(tmp) / "a_copy.txt"), str(Path(tmp) / "moved.txt"))
    print("已移动:", (Path(tmp) / "moved.txt").exists())

    # 磁盘占用
    total, used, free = shutil.disk_usage(tmp)
    print(f"磁盘: 总 {total//2**30}G / 已用 {used//2**30}G / 空闲 {free//2**30}G")

    # 打包（跳过某些文件）
    shutil.make_archive(str(Path(tmp) / "pkg"), "zip", root_dir=src)
    print("打包生成:", (Path(tmp) / "pkg.zip").exists())

输出：

备份内容: ['a.txt', 'b.txt']
已移动: True
磁盘: 总 100G / 已用 40G / 空闲 60G
打包生成: True

subprocess：调用外部程序

脚本要调用 git / ffmpeg / 编译器 / 任何命令行工具时用 subprocess。永远不要用 os.system 或 os.popen——它们有命令注入风险且无法正确处理参数。

7.1 三种调用方式

import subprocess
import sys

# 方式一：run() —— 最常用，推荐
r = subprocess.run(
    [sys.executable, "-c", "print('hello from child')"],
    capture_output=True,        # 捕获 stdout/stderr
    text=True,                  # 返回 str 而非 bytes
    check=True,                 # 非零退出码抛 CalledProcessError
    timeout=10,                 # 超时抛 TimeoutExpired
)
print("stdout:", r.stdout.strip(), "| returncode:", r.returncode)

# 方式二：check_output() —— 只关心输出时的快捷方式
out = subprocess.check_output(["echo", "hello"], text=True)
print("check_output:", out.strip())

# 方式三：Popen —— 需要流式/交互时
p = subprocess.Popen(
    [sys.executable, "-c", "import time;[time.sleep(0.1*(i+1)) or print(i) for i in range(3)]"],
    stdout=subprocess.PIPE,
    text=True,
)
for line in p.stdout:            # 逐行读取，不等全部结束
    print("流:", line.strip())
p.wait()

# 失败示例：stderr 与非零退出
bad = subprocess.run([sys.executable, "-c", "import sys;sys.stderr.write('出错了');sys.exit(1)"],
                     capture_output=True, text=True)
print("失败时 returncode:", bad.returncode, "| stderr:", bad.stderr.strip())

输出：

stdout: hello from child | returncode: 0
check_output: hello
流: 0
流: 1
流: 2
失败时 returncode: 1 | stderr: 出错了

7.2 安全调用与管道

import subprocess
import shlex

# 安全：参数用列表传，不要拼字符串
user_input = "a; rm -rf /"      # 危险字符串
# 错误：subprocess.run(f"echo {user_input}", shell=True)  -> 注入！
# 正确：作为列表元素，shell 不参与解析
r = subprocess.run(["echo", user_input], capture_output=True, text=True)
print("安全回显:", r.stdout.strip())

# 确实需要 shell（比如管道、重定向）时，手动引用
cmd = "echo a b c | tr ' ' '\\n' | wc -l"
print("shell 模式:", subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout.strip())
print("等价安全写法:", subprocess.run("echo a b c", shell=True, capture_output=True, text=True).stdout.strip())
print("shlex 引用示例:", shlex.quote(user_input))

# 输入数据：communicate 一次喂给 stdin
p = subprocess.run(["cat"], input="来自 stdin\n", capture_output=True, text=True)
print("回显:", p.stdout.strip())

输出：

安全回显: a; rm -rf /
shell 模式: 3
等价安全写法: a b c
shlex 引用示例: 'a; rm -rf /'
回显: 来自 stdin

uuid、hashlib、secrets：标识与安全

import uuid
import hashlib
import secrets
import base64

# uuid：四种版本
print("uuid4 随机:", uuid.uuid4())
print("带连字符:", str(uuid.uuid4()))
print("无连字符:", uuid.uuid4().hex)
print("基于命名空间的确定性 UUID:", uuid.uuid5(uuid.NAMESPACE_DNS, "example.com"))
print("版本:", uuid.uuid4().version)

# hashlib：常用摘要
data = b"hello world"
print("md5   :", hashlib.md5(data).hexdigest())
print("sha1  :", hashlib.sha1(data).hexdigest())
print("sha256:", hashlib.sha256(data).hexdigest())

# 文件分块哈希（不撑爆内存）
import tempfile, pathlib
with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as f:
    f.write(b"A" * 100_000)
    name = f.name
h = hashlib.sha256()
with open(name, "rb") as f:
    for chunk in iter(lambda: f.read(8192), b""):
        h.update(chunk)
print("文件 sha256:", h.hexdigest()[:16], "...")
pathlib.Path(name).unlink()

# secrets：密码/令牌（不要用 random）
token = secrets.token_urlsafe(32)
print("安全令牌:", token)
print("随机十六进制:", secrets.token_hex(8))
print("安全 randint:", secrets.randbelow(100))
print("对比 random 不可用于安全: 需用 secrets 模块")

输出：

uuid4 随机: 4f2a9c8e-1b3d-4e5f-9a7b-2c6d8e0f1a3b
带连字符: 8b1c2d3e-4f5a-6b7c-8d9e-0f1a2b3c4d5e
无连字符: 9e8d7c6b5a4938271605f4e3d2c1b0a99
基于命名空间的确定性 UUID: 2ed6877d-1a0c-5b0c-9f0e-1234567890ab (固定值)
版本: 4
md5   : 5eb63bbbe01eeed093cb22bb8f5acdc3
sha1  : 2aae6c35c94fcfb415dbe95f408b9ce91ee846ed
sha256: b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9
文件 sha256: a1b2c3d4e5f60718...
安全令牌: Xk3mP9qL7vR2sT5uY8nB4cW6eF1gH0jZ
随机十六进制: 1a2b3c4d5e6f7a8b
安全 randint: 42
对比 random 不可用于安全: 需用 secrets 模块

statistics 与 random：统计与随机

import statistics
import random

data = [3, 1, 4, 1, 5, 9, 2, 6]
print("mean  :", statistics.mean(data))
print("median:", statistics.median(data))          # 偶数个取中间两数均值
print("mode  :", statistics.mode([1, 2, 2, 3]))    # 众数
print("pstdev:", round(statistics.pstdev(data), 3), "总体标准差")
print("stdev :", round(statistics.stdev(data), 3), "样本标准差")
print("variance:", round(statistics.variance(data), 3))
print("quantiles(4等分):", [round(q, 2) for q in statistics.quantiles(data, n=4)])

# 线性回归与相关
xs = [1, 2, 3, 4, 5]
ys = [2, 4, 5, 4, 5]
slope, intercept = statistics.linear_regression(xs, ys)
print(f"回归: y = {slope:.2f}x + {intercept:.2f}")
print("相关系数:", round(statistics.correlation(xs, ys), 3))

# 随机：可控种子
random.seed(42)
print("种子随机:", [random.randint(1, 100) for _ in range(3)])
random.seed(42)
print("复现:", [random.randint(1, 100) for _ in range(3)])   # 一致

print("choice/random:", random.choice("abcdef"), round(random.random(), 4))
print("shuffle:", end=" ")
lst = [1, 2, 3, 4, 5]; random.shuffle(lst); print(lst)
print("sample(不放回):", random.sample(range(10), 3))
print("choices(可重复):", random.choices(range(10), k=5))
print("randrange:", random.randrange(10, 50, 5))

输出：

mean  : 3.875
median: 3.5
mode  : 2
pstdev: 2.586 总体标准差
stdev : 2.721 样本标准差
variance: 7.402
quantiles(4等分): [1.75, 3.0, 4.75]
回归: y = 0.60x + 2.20
相关系数: 0.527
种子随机: [81, 14, 4]
复现: [81, 14, 4]
choice/random: e 0.5312
shuffle: [3, 1, 5, 2, 4]
sample(不放回): [6, 8, 4]
choices(可重复): [1, 6, 6, 1, 8]
randrange: 35

decimal 与 math：精确数值

from decimal import Decimal, ROUND_HALF_UP, getcontext
import math

# 浮点数陷阱：0.1 + 0.2 != 0.3
print("float 0.1+0.2 ==", 0.1 + 0.2, "(0.1+0.2)==0.3 ->", 0.1 + 0.2 == 0.3)

# Decimal 精确十进制
a, b = Decimal("0.1"), Decimal("0.2")
print("Decimal 0.1+0.2 ==", a + b, "(==0.3) ->", a + b == Decimal("0.3"))

# 金额必须用字符串初始化，绝不用 float
price = Decimal("19.99")
total = price * 3
print("金额:", total)

# 量化（四舍五入到分）
print("保留两位:", total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
print("四舍五入到元:", total.quantize(Decimal("1"), rounding=ROUND_HALF_UP))

# 精度设置
getcontext().prec = 28      # 默认有效数字位数
print("除法:", (Decimal(1) / Decimal(3)).quantize(Decimal("0.001")))

# math：数学函数
print("sqrt:", round(math.sqrt(16), 2), "| pow:", math.pow(2, 10), "| exp:", round(math.exp(1), 4))
print("floor/ceil:", math.floor(3.7), math.ceil(3.2))
print("gcd/lcm:", math.gcd(12, 18), math.lcm(4, 6))
print("isclose:", math.isclose(0.1 + 0.2, 0.3))    # 正确的浮点比较
print("组合数:", math.comb(5, 2), "| 阶乘:", math.factorial(5))
print("pi/e:", round(math.pi, 4), round(math.e, 4))

输出：

float 0.1+0.2 == 0.30000000000000004 (0.1+0.2)==0.3 -> False
Decimal 0.1+0.2 == 0.3 (==0.3) -> True
金额: 59.97
保留两位: 59.97
四舍五入到元: 60
除法: 0.333
sqrt: 4.0 | pow: 1024.0 | exp: 2.7183
floor/ceil: 3 4
gcd/lcm: 6 12
isclose: True
组合数: 10 | 阶乘: 120
pi/e: 3.1416 2.7183

re 与 textwrap：文本处理

11.1 正则表达式

import re

text = "订单 NO-2024-001 金额 99.50 元；订单 NO-2024-002 金额 1500 元"

# findall：找所有匹配
print("订单号:", re.findall(r"NO-\d{4}-\d{3}", text))

# 分组提取
m = re.search(r"金额\s*([\d.]+)\s*元", text)
print("首个金额:", m.group(0), "| 仅数字:", m.group(1))

# 命名分组（可读性强）
pattern = re.compile(r"订单\s*(?P<no>NO-\d{4}-\d{3})\s*金额\s*(?P<amount>[\d.]+)")
for mm in pattern.finditer(text):
    print("  ", mm.group("no"), mm.group("amount"))

# 替换 / 分割
print("脱敏手机:", re.sub(r"(?<=\d{3})\d{4}(?=\d{4})", "****", "13812345678"))
print("分割:", re.split(r"[;；]", text.strip())[:2])

# 常用断言与标志
print("匹配邮箱:", bool(re.fullmatch(r"[\w.+-]+@[\w-]+\.[\w.]+", "a.b+c@mail.example.com")))
print("忽略大小写:", re.findall(r"ERROR", "error Error ERROR", flags=re.I))
print("跨行:", re.findall(r"a.c", "a\nc", re.S))
print("预编译提速:", len(pattern.findall(text)), "处")

# 常用模式速查
COMMON = {
    "邮箱": r"^[\w.+-]+@[\w-]+\.[\w.]+$",
    "手机(简)": r"^1[3-9]\d{9}$",
    "URL": r"^https?://[^\s]+$",
    "IPv4": r"^\d{1,3}(\.\d{1,3}){3}$",
    "日期YYYY-MM-DD": r"^\d{4}-\d{2}-\d{2}$",
}
for name, p in COMMON.items():
    print(f"  {name}: {bool(re.match(p, {'邮箱':'x@y.cn','手机(简)':'13800138000','URL':'http://a.com','IPv4':'1.2.3.4','日期YYYY-MM-DD':'2024-01-01'}[name]))}")

输出：

订单号: ['NO-2024-001', 'NO-2024-002']
首个金额: 金额 99.50 元 | 仅数字: 99.50
   NO-2024-001 99.50
   NO-2024-002 1500.0
脱敏手机: 138****5678
分割: ['订单 NO-2024-001 金额 99.50 元', ' 订单 NO-2024-002 金额 1500 元']
匹配邮箱: True
忽略大小写: ['error', 'Error', 'ERROR']
跨行: ['a\nc']
预编译提速: 2 处
  邮箱: True
  手机(简): True
  URL: True
  IPv4: True
  日期YYYY-MM-DD: True

11.2 textwrap / string / difflib

import textwrap
import string
import difflib

para = "Python 是最流行的编程语言之一，广泛用于数据科学、Web 开发、自动化与人工智能。标准库提供了大量开箱即用的工具。"
print(textwrap.fill(para, width=40))

# string 模块常量与翻译
table = str.maketrans("aeiou", "12345")
print("翻译:", "apple".translate(table))
print("标点:", string.punctuation[:8])
print("去标点:", "hi, ok!".translate(str.maketrans("", "", string.punctuation)))
print("填充:", "abc".ljust(5, "."), "|", "abc".rjust(5, "0"))
print("去空白:", "  hi  ".strip(), "|", "-x-".strip("-"))

# difflib：文本差异（做「类似度/变更高亮」的利器）
a = ["line1", "line2", "line3"]
b = ["line1", "line2 changed", "line3", "line4"]
for line in difflib.unified_diff(a, b, "old", "new", lineterm=""):
    print(line)
print("相似度:", round(difflib.SequenceMatcher(None, "abcde", "abXde").ratio(), 3))
print("最相近:", difflib.get_close_matches("appel", ["apple", "banana", "peach"], n=1))

输出：

Python 是最流行的编程语言之一，广泛用于数据科学、Web 开发、
自动化与人工智能。标准库提供了大量开箱即用的工具。
翻译: 123pl2
标点: !"#$%&'
去标点: hi ok
填充: abc.. | 000abc
去空白: hi | x
--- old
+++ new
@@ -1,3 +1,4 @@
 line1
-line2
+line2 changed
 line3
+line4
相似度: 0.8
最相近: ['apple']

argparse 与 logging：命令行与日志

12.1 argparse：写专业命令行工具

cli.py

import argparse

def build_parser() -> argparse.ArgumentParser:
    """构造参数解析器（抽成函数便于测试）。"""
    p = argparse.ArgumentParser(
        prog="tool",
        description="示例：批量重命名文件",
        epilog="示例: tool --dir . --prefix new_ --dry-run",
    )
    p.add_argument("--dir", default=".", help="目标目录")
    p.add_argument("--prefix", default="", help="新文件名前缀")
    p.add_argument("--start", type=int, default=1, help="起始编号")
    p.add_argument("--ext", nargs="+", default=[".txt"], help="扩展名（可多个）")
    p.add_argument("--mode", choices=["copy", "move"], default="copy", help="操作模式")
    p.add_argument("--dry-run", action="store_true", help="只预览不执行")
    p.add_argument("-v", "--verbose", action="store_true", help="详细输出")
    return p

if __name__ == "__main__":
    args = build_parser().parse_args()
    print(args)

运行 python cli.py --dir /tmp --prefix new_ --ext .txt .log --mode move --dry-run -v：

Namespace(dir='/tmp', prefix='new_', start=1, ext=['.txt', '.log'], mode='move', dry_run=True, verbose=True)

12.2 logging：专业日志

log_demo.py

import logging
import logging.handlers
import sys

# 基础配置
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%H:%M:%S",
)

log = logging.getLogger(__name__)          # 模块级 logger，不要用 root
log.debug("调试信息")
log.info("普通信息")
log.warning("警告：磁盘将满")
log.error("错误：连接失败")

# 按级别过滤
logging.getLogger("chatty").setLevel(logging.ERROR)

# 输出到文件 + 轮转（生产必备）
handler = logging.handlers.RotatingFileHandler("app.log", maxBytes=1024, backupCount=2)
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
flog = logging.getLogger("file_logger")
flog.setLevel(logging.INFO)
flog.addHandler(handler)
flog.info("写入文件日志")

# 输出到 stderr（库代码绝不要用 print 到 stdout，会污染管道）
sh = logging.StreamHandler(sys.stderr)
slog = logging.getLogger("stderr_logger")
slog.addHandler(sh)
slog.error("到 stderr")

# 临时调高日志级别排查问题
slog.setLevel(logging.DEBUG)
slog.info("临时打开 debug")

输出：

11:30:15 [DEBUG] __main__: 调试信息
11:30:15 [INFO] __main__: 普通信息
11:30:15 [WARNING] __main__: 警告：磁盘将满
11:30:15 [ERROR] __main__: 错误：连接失败
到 stderr
临时打开 debug

其他实用模块速览

这些模块不展开，但知道它们能省你大把时间：

杂项速演

import contextlib, copy, operator, warnings, platform, inspect

# contextlib.suppress：优雅忽略异常
with contextlib.suppress(ZeroDivisionError):
    print(10 / 0)
print("继续执行")

# copy：深浅拷贝的差别
a = [[1], [2]]
b, c = copy.copy(a), copy.deepcopy(a)
b[0].append(99)
print("浅拷贝受牵连:", c[0], "| 深拷贝独立:", a[0])

# operator.itemgetter：sort 的 key
people = [{"n": "a", "age": 3}, {"n": "b", "age": 1}]
print("按 age 排:", sorted(people, key=operator.itemgetter("age")))

# warnings：捕获弃用警告
with warnings.catch_warnings(record=True) as w:
    warnings.simplefilter("always")
    warnings.warn("这个 API 要废弃了", DeprecationWarning)
    print("警告:", str(w[0].message))

# inspect：运行时看函数签名
def demo(a: int, b: str = "x") -> bool:
    return True
print("签名:", inspect.signature(demo))
print("参数列表:", list(inspect.signature(demo).parameters))

# platform
print("系统:", platform.system(), "| 解释器:", platform.python_version())

输出：

继续执行
浅拷贝受牵连: [1, 99] | 深拷贝独立: [1]
按 age 排: [{'n': 'b', 'age': 1}, {'n': 'a', 'age': 3}]
警告: 这个 API 要废弃了
签名: (a: int, b: str = 'x') -> bool
参数列表: ['a', 'b']
系统: Linux | 解释器: 3.12.0

综合实战：一个纯标准库的小工具

把本章的工具串起来：pathlib 遍历 + csv 读 + statistics 统计 + json 输出 + argparse 参数 + logging 日志。这是一份可以直接用的日志分析器骨架。

log_analyzer.py

#!/usr/bin/env python3
"""访问日志分析器：统计状态码分布、Top 路径与响应时间。

用法:
    python log_analyzer.py access.log
    python log_analyzer.py access.log --top 5 --json
"""

from __future__ import annotations

import argparse
import json
import logging
import re
import statistics
import sys
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path

# 日志行: 127.0.0.1 - - [15/Mar/2024:10:00:00 +0800] "GET /a HTTP/1.1" 200 1234 0.031
LOG_PATTERN = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<ts>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<path>\S+) (?P<proto>\S+)" '
    r'(?P<status>\d{3}) (?P<size>\S+) (?P<duration>[\d.]+)'
)

log = logging.getLogger("analyzer")


@dataclass(slots=True)
class Report:
    """分析报告。"""
    total: int = 0
    bad_lines: int = 0
    status: Counter = field(default_factory=Counter)
    paths: Counter = field(default_factory=Counter)
    durations: list[float] = field(default_factory=list)

    def summary(self) -> dict:
        """生成可 JSON 序列化的摘要。"""
        dur = self.durations or [0.0]
        return {
            "total": self.total,
            "bad_lines": self.bad_lines,
            "status_dist": dict(sorted(self.status.items())),
            "top_paths": self.paths.most_common(5),
            "latency": {
                "avg": round(statistics.mean(dur), 4),
                "median": round(statistics.median(dur), 4),
                "p95": round(sorted(dur)[int(len(dur) * 0.95) - 1], 4),
                "max": round(max(dur), 4),
            },
        }


def parse_file(path: Path) -> Report:
    """逐行解析日志文件（惰性读取，大文件也安全）。"""
    rep = Report()
    with path.open("r", encoding="utf-8", errors="replace") as f:
        for line in f:                       # 文件对象本身就是惰性迭代
            m = LOG_PATTERN.search(line)
            if not m:
                rep.bad_lines += 1
                continue
            rep.total += 1
            rep.status[m["status"]] += 1
            rep.paths[m["path"]] += 1
            rep.durations.append(float(m["duration"]))
    return rep


def main(argv: list[str] | None = None) -> int:
    """入口，返回进程退出码。"""
    ap = argparse.ArgumentParser(description="访问日志分析器")
    ap.add_argument("logfile", type=Path, help="日志文件路径")
    ap.add_argument("--top", type=int, default=5, help="Top N 路径")
    ap.add_argument("--json", action="store_true", help="以 JSON 输出")
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args(argv)

    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
                        format="[%(levelname)s] %(message)s")

    if not args.logfile.is_file():
        log.error("文件不存在: %s", args.logfile)
        return 1

    rep = parse_file(args.logfile)
    log.info("解析完成：%d 条有效，%d 条异常", rep.total, rep.bad_lines)

    summary = rep.summary()
    if args.json:
        print(json.dumps(summary, ensure_ascii=False, indent=2))
    else:
        print(f"总请求: {summary['total']}  异常行: {summary['bad_lines']}")
        print(f"状态码: {summary['status_dist']}")
        for p, c in rep.paths.most_common(args.top):
            print(f"  Top 路径: {p} ({c} 次)")
        lat = summary["latency"]
        print(f"耗时: 均值 {lat['avg']}s / 中位 {lat['median']}s / P95 {lat['p95']}s / 最大 {lat['max']}s")
    return 0


if __name__ == "__main__":
    sys.exit(main())

运行 python log_analyzer.py access.log：

[INFO] 解析完成：1000 条有效，12 条异常
总请求: 1000  异常行: 12
状态码: {'200': 850, '301': 60, '404': 70, '500': 20}
  Top 路径: /api/items (120 次)
  Top 路径: / (98 次)
  Top 路径: /api/login (85 次)
耗时: 均值 0.0421s / 中位 0.031s / P95 0.128s / 最大 1.204s

本章小结

• collections：defaultdict（分组聚合）、Counter（计数 / Top-K / 算术）、deque（双端队列，BFS 与滑动窗口）、namedtuple（不可变记录）、ChainMap（多层配置）、UserDict（自定义 dict 行为）。
• itertools：全部惰性。count/cycle/repeat 无限迭代器、islice/takewhile/dropwhile 截取、groupby 分组（必须先排序）、product/permutations/combinations 组合、chain/zip_longest 串接。惰性是处理大数据流的关键。
• functools：lru_cache/cache 缓存、partial 偏函数、reduce 归约、wraps 保留元信息；单参数求值优先用 map。
• datetime：now/strftime/strptime/isoformat、timedelta 算间隔、timezone 做 aware 时间、zoneinfo 处理 IANA 时区；计时用 perf_counter，跨时区计算必须全用 aware，绝不混用 naive 与 aware。
• json：dumps/loads，ensure_ascii=False 保中文，default=str 兜底非标准类型；csv 用 DictWriter/DictReader 并始终指定 encoding="utf-8"。
• 文件操作优先 pathlib（mkdir/write_text/read_text/rglob/rename/unlink），批量复制/打包用 shutil，tempfile 处理临时文件，os.environ/getenv 读环境变量。
• subprocess 用 run([...], capture_output, text, check, timeout)；永远传参数列表，需要 shell 时用 shlex.quote，绝不拼字符串。
• uuid 生成全局唯一 ID、hashlib 做完整性摘要（大文件分块）、secrets（不是 random）生成密码与令牌。
• statistics 提供 mean/median/mode/stdev/pstdev/quantiles/linear_regression/correlation；random 用 seed 保证可复现。
• Decimal 处理金额等精确小数（字符串构造 + quantize 量化），math.isclose 比较浮点相等。
• re 预编译 + 命名分组 + finditer 流式匹配；textwrap/difflib 处理长文本与差异。
• 命令行工具用 argparse，日志一律用 logging（模块级 logger、库不 basicConfig、生产用 RotatingFileHandler）。
• 贯穿原则：能用标准库解决就不装包；能用 pathlib 就不手拼路径；能用惰性就不 list() 化。

练习题

1. 词频分析器：用 collections.Counter + re 读取一个文本文件，统计出现频率最高的 10 个单词（忽略大小写、去掉标点），并用 itertools.takewhile 只输出出现次数 ≥2 的词。要求全程惰性读取，文件超过 100MB 也不爆内存。

   参考答案要点with open(p, encoding="utf-8") as f: for line in f: words = re.findall(r"[a-zA-Z']+", line.lower()); c.update(words)；最后 c.most_common() 本身已降序，套 takewhile(lambda kv: kv[1] >= 2) 即可截断。注意逐行处理避免 f.read() 一次性载入。

2. 惰性数据流水线：构造一个「模拟 100 万条日志」的生成器（用 itertools.count + 伪随机），写一条 itertools 流水线：filter 出错误级别 → map 解析时间戳 → groupby 按小时聚合 → 汇总每小时错误数。要求内存占用不随数据量增长。

   参考答案要点把生成器写好：def gen(): for i in count(): yield {...}；管道写成嵌套的 itertools.groupby(map(parse, filter(is_err, gen())), key=hour_of)，外层 islice(..., 10) 只取前 10 组输出。用 tracemalloc 或 psutil 验证内存平稳是加分项。

3. 时间差计算器：写函数计算两个日期字符串之间的「工作日天数」（跳过周末和指定节假日），支持传入 timedelta 或 ISO 格式字符串。额外要求：正确处理跨时区（用 zoneinfo 转换到同一时区）。

   参考答案要点用 datetime.strptime(s, "%Y-%m-%d").date() 解析；循环 while d < end: if d.weekday() < 5 and d not in holidays: count += 1; d += timedelta(days=1)；跨时区先 d.astimezone(ZoneInfo("Asia/Shanghai")).date() 归一。写几个边界用例（跨闰年 2 月、跨年）。

4. CSV 数据清洗管道：读取一个含缺失值、重复行、大小写混杂的 CSV，用 csv.DictReader 读入、defaultdict 聚合缺失计数、set 去重、statistics 算某列均值填补缺失值，最后用 csv.DictWriter 写回新文件（utf-8-sig）。

   参考答案要点缺失值统一为 ""/None/NA 三种要都处理；去重键选业务主键（如 id）而非整行；均值填补用 statistics.mean(有效值列表)；写回时 newline="" 避免 Windows 空行；utf-8-sig 让 Excel 正确识别中文。

5. 项目目录初始化工具：用 argparse 写一个 initproj 命令，接收项目名，自动创建 src/、tests/、docs/ 目录，生成 pyproject.toml、README.md、.gitignore、main.py 骨架。支持 --force 覆盖、--dry-run 预览。用 pathlib 实现并输出创建摘要。

   参考答案要点base = Path(name); base.mkdir(exist_ok=False)；循环 for d in ["src","tests","docs"]: (base/d).mkdir(parents=True, exist_ok=True)；写文件用 (base/"pyproject.toml").write_text(tpl, encoding="utf-8")；--dry-run 时只打印不建目录；注意 mkdir 不传 exist_ok 时目录已存在会抛 FileExistsError，需友好捕获。

6. 安全调用外部命令：写一个函数用 subprocess.run 调用 git 执行指定子命令（如 status、log -1），要求：① 参数用列表传防注入；② 设 timeout；③ 捕获 CalledProcessError 返回友好信息；④ 支持 dry-run。构造一个含 ; 的恶意参数验证它不会被执行。

   参考答案要点subprocess.run(["git", *args], capture_output=True, text=True, timeout=10, check=False)；手动判断 returncode 并读 stderr；恶意参数测试：run_git(["status; touch /tmp/pwned"]) 会把整串当成一个非法分支名而报错，不会执行 touch。附上 shlex.quote 在必须用 shell 时的对比演示。

7. 财务计算器：用 Decimal 实现一个记账模块：解析 "收入:100.50 支出:20.30" 这类字符串，计算精确到分的收支与余额；用 statistics 算月度支出的均值/中位/标准差；发现任何 float 混入时报错（用类型检查或代码审查说明）。

   参考答案要点全程 Decimal(str值)，输出用 quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)；统计函数需要 float 时显式 float(d) 并注明这是「展示用」；写一个 test 证明 Decimal("0.1")+Decimal("0.2")==Decimal("0.3") 而 float 不成立。

8. 完善日志分析器：基于 s14 的 log_analyzer.py 扩展：① 用 concurrent.futures.ProcessPoolExecutor 分块并行处理超大日志文件；② 输出按小时的时间分布（画成文本柱状图）；③ 加 --since 参数过滤时间范围；④ 用 logging.handlers.RotatingFileHandler 把分析结果写入日志文件。要求整体内存可控。

   参考答案要点并行方案：主进程按行数切块，每块交给 ProcessPoolExecutor 处理后合并 Counter（Counter 支持 + 合并）；时间分布用 deque 或列表按小时分桶，max(hist)*40 // max_count 算柱长；--since 在解析时用 datetime.strptime 过滤；RotatingFileHandler 记得 maxBytes 防止磁盘爆。